GO-3354 subscription publish undetermined order

pkg/lib/database/subscription.go

@@ -1,20 +1,25 @@
 package database
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
 
+	"github.com/cheggaaa/mb/v3"
 	"github.com/gogo/protobuf/types"
 	"golang.org/x/exp/slices"
 )
 
 type subscription struct {
-	ids    []string
-	quit   chan struct{}
-	closed bool
-	ch     chan *types.Struct
-	wg     sync.WaitGroup
+	ids              []string
+	quit             chan struct{}
+	closed           bool
+	ch               chan *types.Struct
+	wg               sync.WaitGroup
+	publishQueue     mb.MB[*types.Struct]
+	processQueueOnce sync.Once
 	sync.RWMutex
 }
 
@@ -23,7 +28,12 @@
 	RecordChan() chan *types.Struct
 	Subscribe(ids []string) (added []string)
 	Subscriptions() []string
+	// Publish is blocking
+	// returns false if the subscription is closed or the id is not subscribed
 	Publish(id string, msg *types.Struct) bool
+	// PublishAsync is non-blocking and guarantees the order of messages
+	// returns false if the subscription is closed or the id is not subscribed
+	PublishAsync(id string, msg *types.Struct) bool
 }
 
 func (sub *subscription) RecordChan() chan *types.Struct {
@@ -40,6 +50,7 @@
 	sub.Unlock()
 	close(sub.quit)
 
+	sub.publishQueue.Close()
 	sub.wg.Wait()
 	close(sub.ch)
 }
@@ -60,6 +71,63 @@
 	return
 }
 
+// should be called via sub.processQueueOnce
+func (sub *subscription) processQueue() {
+	go func() {
+		select {
+		case <-sub.quit:
+			err := sub.publishQueue.Close()
+			if err != nil && !errors.Is(err, mb.ErrClosed) {
+				log.Errorf("subscription %p failed to close async queue: %s", sub, err)
+			}
+			unprocessed := sub.publishQueue.Len()
+			if unprocessed > 0 {
+				log.Errorf("subscription %p has %d unprocessed messages in the async queue", sub, unprocessed)
+			}
+		}
+	}()
+
+	var (
+		msg *types.Struct
+		err error
+	)
+	for {
+		// no need for cancellation here, because we close the queue itself on quit and it will return
+		msg, err = sub.publishQueue.WaitOne(context.Background())
+		if err != nil {
+			if !errors.Is(err, mb.ErrClosed) {
+				log.Errorf("subscription %p failed to get message from async queue: %s", sub, err)
+			}
+			return
+		}
+		select {
+		case sub.ch <- msg:
+			continue
+		}
+	}
+}
+
+// PublishAsync is non-blocking and guarantees the order of messages
+// returns false if the subscription is closed or the id is not subscribed
+func (sub *subscription) PublishAsync(id string, msg *types.Struct) bool {
+	sub.RLock()
+	if sub.closed {
+		sub.RUnlock()
+		return false
+	}
+	if !slices.Contains(sub.ids, id) {
+		sub.RUnlock()
+		return false
+	}
+	sub.RUnlock()
+	sub.processQueueOnce.Do(func() {
+		go sub.processQueue()
+	})
+	log.Debugf("objStore subscription sendasync %s %p", id, sub)
+	err := sub.publishQueue.Add(context.Background(), msg)
+	return err == nil
+}
+
 func (sub *subscription) Publish(id string, msg *types.Struct) bool {
 	sub.RLock()
 	if sub.closed {

pkg/lib/localstore/objectstore/queries_test.go

@@ -3,6 +3,7 @@ package objectstore
 import (
 	"fmt"
 	"sort"
+	"strconv"
 	"testing"
 	"time"
 
@@ -647,15 +648,27 @@ func TestQueryByIdAndSubscribeForChanges(t *testing.T) {
 		}
 	})
 
-	t.Run("update details", func(t *testing.T) {
-		err = s.UpdateObjectDetails("id1", makeDetails(makeObjectWithName("id1", "name1 updated")))
-		require.NoError(t, err)
+	t.Run("update details order", func(t *testing.T) {
+		for i := 1; i <= 1000; i++ {
+			err = s.UpdateObjectDetails("id1", makeDetails(makeObjectWithName("id1", fmt.Sprintf("%d", i))))
+			require.NoError(t, err)
+		}
 
-		select {
-		case rec := <-recordsCh:
-			assert.Equal(t, "name1 updated", pbtypes.GetString(rec, bundle.RelationKeyName.String()))
-		case <-time.After(10 * time.Millisecond):
-			require.Fail(t, "update has not been received")
+		prev := 0
+		for {
+			select {
+			case rec := <-recordsCh:
+				name := pbtypes.GetString(rec, bundle.RelationKeyName.String())
+				num, err := strconv.Atoi(name)
+				require.NoError(t, err)
+				require.Equal(t, prev+1, num)
+				if num == 1000 {
+					return
+				}
+				prev = num
+			case <-time.After(10 * time.Millisecond):
+				require.Fail(t, "update has not been received")
+			}
 		}
 	})
 }

pkg/lib/localstore/objectstore/update.go

@@ -208,9 +208,7 @@ func (s *dsObjectStore) sendUpdatesToSubscriptions(id string, details *types.Str
 			Details: detCopy,
 		})
 	}
-	for i := range s.subscriptions {
-		go func(sub database.Subscription) {
-			_ = sub.Publish(id, detCopy)
-		}(s.subscriptions[i])
+	for _, sub := range s.subscriptions {
+		_ = sub.PublishAsync(id, detCopy)
 	}
 }