DSET-1556: Add test to check that each event is processed once (#12)

* DSET-1556: Add test to check that each event is processed once

Author: martin-majlis-s1

.gitignore

@@ -12,4 +12,10 @@
 *.out
 
 # Dependency directories (remove the comment below to include it)
-# vendor/
+vendor/
+
+# log files
+out*.log
+
+.idea
+coverage*

Makefile

@@ -24,9 +24,9 @@ endif
 
 GO_BUILD_TAGS=""
 GOTEST_OPT?= -race -timeout 300s -parallel 4 -count=1 --tags=$(GO_BUILD_TAGS)
-GOTEST_INTEGRATION_OPT?= -race -timeout 360s -parallel 4
+GOTEST_LONG_RUNNING_OPT?= -race -timeout 360s -parallel 4 -count=1 -tags=long_running,$(GO_BUILD_TAGS)
 GOTEST_OPT_WITH_COVERAGE = $(GOTEST_OPT) -coverprofile=coverage.txt -covermode=atomic
-GOTEST_OPT_WITH_INTEGRATION=$(GOTEST_INTEGRATION_OPT) -tags=integration,$(GO_BUILD_TAGS) -run=Integration -coverprofile=integration-coverage.txt -covermode=atomic
+GOTEST_OPT_WITH_COVERAGE_LONG_RUNNING=$(GOTEST_LONG_RUNNING_OPT) -coverprofile=coverage.txt -covermode=atomic
 GOCMD?= go
 GOTEST=$(GOCMD) test
 
@@ -45,9 +45,16 @@ build:
 	echo "Done"
 
 .PHONY: test
-test:
+test: test-all
+
+.PHONY: test-unit
+test-unit:
 	$(GOTEST) $(GOTEST_OPT) ./...
 
+.PHONY: test-all
+test-all:
+	$(GOTEST) $(GOTEST_LONG_RUNNING_OPT) ./...
+
 .PHONY: test-many-times
 test-many-times:
 	set -e; \
@@ -58,11 +65,20 @@ test-many-times:
   	fi; \
   	for i in `seq 1 $${COUNT}`; do \
   		echo "Running test $${i} / $${COUNT}"; \
-  		make test; \
+  		make test 2>&1 | tee out-test-$${i}.log; \
   	done;
 
 
 .PHONY: coverage
-coverage:
+coverage: coverage-all
+
+.PHONY: coverage-unit
+coverage-unit:
 	$(GOTEST) $(GOTEST_OPT_WITH_COVERAGE) ./...
 	$(GOCMD) tool cover -html=coverage.txt -o coverage.html
+
+
+.PHONY: coverage-all
+coverage-all:
+	$(GOTEST) $(GOTEST_OPT_WITH_COVERAGE_LONG_RUNNING) ./...
+	$(GOCMD) tool cover -html=coverage.txt -o coverage.html

RELEASE_NOTES.md

@@ -1,5 +1,9 @@
 # Release Notes
 
+## 0.0.4 Fix Concurrency Issues
+
+* sometimes not all events have been delivered exactly once
+
 ## 0.0.3 Fix Data Races
 
 * fixed [data races](https://go.dev/doc/articles/race_detector)

pkg/buffer/buffer.go

@@ -19,6 +19,7 @@ package buffer
 import (
 	"encoding/json"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -41,12 +42,17 @@ type Status uint32
 const (
 	Initialising = Status(iota)
 	Ready
+	AddingBundles
 	Publishing
 	Retrying
 )
 
 func (s Status) String() string {
-	return [...]string{"Initialising", "Ready", "Publishing", "Retrying"}[s]
+	return [...]string{"Initialising", "Ready", "AddingBundles", "Publishing", "Retrying"}[s]
+}
+
+func (s Status) IsActive() bool {
+	return s == Ready || s == AddingBundles
 }
 
 type AddStatus uint8
@@ -74,14 +80,16 @@ type Buffer struct {
 	Session string
 	Token   string
 
-	Status    atomic.Uint32
-	Attempt   uint
-	createdAt atomic.Int64
+	Attempt     uint
+	createdAt   atomic.Int64
+	status      atomic.Uint32
+	PublishAsap atomic.Bool
 
 	sessionInfo *add_events.SessionInfo
 	threads     map[string]*add_events.Thread
 	logs        map[string]*add_events.Log
 	events      []*add_events.Event
+	dataMutex   sync.Mutex
 
 	lenSessionInfo int
 	lenThreads     atomic.Int32
@@ -100,8 +108,9 @@ func NewEmptyBuffer(session string, token string) *Buffer {
 		Id:           id,
 		Session:      session,
 		Token:        token,
-		Status:       atomic.Uint32{},
+		status:       atomic.Uint32{},
 		Attempt:      0,
+		PublishAsap:  atomic.Bool{},
 		countThreads: atomic.Int32{},
 		countLogs:    atomic.Int32{},
 		countEvents:  atomic.Int32{},
@@ -119,16 +128,17 @@ func NewBuffer(session string, token string, sessionInfo *add_events.SessionInfo
 }
 
 func (buffer *Buffer) Initialise(sessionInfo *add_events.SessionInfo) error {
-	status := Status(buffer.Status.Load())
+	status := buffer.Status()
 	if status != Initialising {
 		panic(fmt.Sprintf("NewBuffer was already initialised: %s", status))
 	}
 	buffer.threads = map[string]*add_events.Thread{}
 	buffer.logs = map[string]*add_events.Log{}
 	buffer.events = []*add_events.Event{}
+	buffer.dataMutex = sync.Mutex{}
 
 	buffer.createdAt.Store(time.Now().UnixNano())
-	buffer.Status.Store(uint32(Ready))
+	buffer.SetStatus(Ready)
 
 	err := buffer.SetSessionInfo(sessionInfo)
 	if err != nil {
@@ -171,8 +181,10 @@ func (buffer *Buffer) SessionInfo() *add_events.SessionInfo {
 }
 
 func (buffer *Buffer) AddBundle(bundle *add_events.EventBundle) (AddStatus, error) {
-	status := Status(buffer.Status.Load())
-	if status != Ready {
+	buffer.dataMutex.Lock()
+	defer buffer.dataMutex.Unlock()
+	status := buffer.Status()
+	if status != Ready && status != AddingBundles {
 		return TooMuch, &NotAcceptingError{status: status}
 	}
 	// append thread
@@ -404,7 +416,7 @@ func (buffer *Buffer) ZapStats(fields ...zap.Field) []zap.Field {
 	res := []zap.Field{
 		zap.String("uuid", buffer.Id.String()),
 		zap.String("session", buffer.Session),
-		zap.Uint32("status", buffer.Status.Load()),
+		zap.String("status", buffer.Status().String()),
 		zap.Uint("attempt", buffer.Attempt),
 		zap.Int32("logs", buffer.countLogs.Load()),
 		zap.Int32("threads", buffer.countThreads.Load()),
@@ -416,3 +428,15 @@ func (buffer *Buffer) ZapStats(fields ...zap.Field) []zap.Field {
 	res = append(res, fields...)
 	return res
 }
+
+func (buffer *Buffer) SetStatus(status Status) {
+	buffer.status.Store(uint32(status))
+}
+
+func (buffer *Buffer) Status() Status {
+	return Status(buffer.status.Load())
+}
+
+func (buffer *Buffer) HasStatus(status Status) bool {
+	return buffer.status.Load() == uint32(status)
+}

pkg/client/add_events.go

@@ -53,48 +53,70 @@ func (client *DataSetClient) AddEvents(bundles []*add_events.EventBundle) error
 	grouped := client.groupBundles(bundles)
 
 	for key, bundles := range grouped {
+		client.addBundle(key, bundles)
+	}
+
+	return nil
+}
 
+func (client *DataSetClient) addBundle(key string, bundles []*add_events.EventBundle) {
+	// this function has to be called from AddEvents - inner loop
+	// it assumes that all bundles have the same key
+	getBuffer := func(key string) *buffer.Buffer {
 		buf := client.Buffer(key, client.SessionInfo)
+		// change state to mark that bundles are being added
+		buf.SetStatus(buffer.AddingBundles)
+		return buf
+	}
+
+	publish := func(key string, buf *buffer.Buffer) *buffer.Buffer {
+		client.PublishBuffer(buf)
+		return getBuffer(key)
+	}
+
+	buf := getBuffer(key)
+	for _, bundle := range bundles {
+		added, err := buf.AddBundle(bundle)
+		if err != nil {
+			if errors.Is(err, &buffer.NotAcceptingError{}) {
+				buf = getBuffer(key)
+			} else {
+				client.Logger.Error("Cannot add bundle", zap.Error(err))
+				// TODO: what to do? For now, lets skip it
+				continue
+			}
+		}
+
+		if buf.ShouldSendSize() || added == buffer.TooMuch && buf.HasEvents() {
+			buf = publish(key, buf)
+		}
 
-		for _, bundle := range bundles {
-			added, err := buf.AddBundle(bundle)
+		if added == buffer.TooMuch {
+			added, err = buf.AddBundle(bundle)
 			if err != nil {
 				if errors.Is(err, &buffer.NotAcceptingError{}) {
-					buf = client.Buffer(key, client.SessionInfo)
+					buf = getBuffer(key)
 				} else {
 					client.Logger.Error("Cannot add bundle", zap.Error(err))
-					// TODO: what to do? For now, lets skip it
 					continue
 				}
 			}
-
-			if buf.ShouldSendSize() || added == buffer.TooMuch && buf.HasEvents() {
-				client.PublishBuffer(buf)
-				buf = client.Buffer(key, client.SessionInfo)
+			if buf.ShouldSendSize() {
+				buf = publish(key, buf)
 			}
-
 			if added == buffer.TooMuch {
-				added, err = buf.AddBundle(bundle)
-				if err != nil {
-					if errors.Is(err, &buffer.NotAcceptingError{}) {
-						buf = client.Buffer(key, client.SessionInfo)
-					} else {
-						client.Logger.Error("Cannot add bundle", zap.Error(err))
-						continue
-					}
-				}
-				if buf.ShouldSendSize() {
-					client.PublishBuffer(buf)
-					buf = client.Buffer(key, client.SessionInfo)
-				}
-				if added == buffer.TooMuch {
-					client.Logger.Fatal("Bundle was not added for second time!", buf.ZapStats()...)
-				}
+				client.Logger.Fatal("Bundle was not added for second time!", buf.ZapStats()...)
 			}
 		}
 	}
+	buf.SetStatus(buffer.Ready)
 
-	return nil
+	// it could happen that the buffer could have been published
+	// by buffer sweeper, but it was skipped, because we have been
+	// adding events, so lets check it and publish it if needed
+	if buf.PublishAsap.Load() {
+		client.PublishBuffer(buf)
+	}
 }
 
 // IsProcessingData returns True if there are still some unprocessed data.
@@ -223,14 +245,19 @@ func (client *DataSetClient) apiCall(req *http.Request, response response.SetRes
 }
 
 func (client *DataSetClient) SendAllAddEventsBuffers() {
+	buffers := client.getBuffers()
 	client.Logger.Debug("Send all AddEvents buffers")
-	client.buffer.Range(func(k, v interface{}) bool {
-		buf, ok := v.(*buffer.Buffer)
-		if ok {
-			client.PublishBuffer(buf)
-		} else {
-			client.Logger.Error("Unable to convert message to buffer")
-		}
-		return true
-	})
+	for _, buf := range buffers {
+		client.PublishBuffer(buf)
+	}
+}
+
+func (client *DataSetClient) getBuffers() []*buffer.Buffer {
+	client.buffersMutex.Lock()
+	defer client.buffersMutex.Unlock()
+	buffers := make([]*buffer.Buffer, 0)
+	for _, buf := range client.buffer {
+		buffers = append(buffers, buf)
+	}
+	return buffers
 }