refactor: Use events to test network logic

cli/start.go

@@ -115,8 +115,6 @@ func MakeStartCommand() *cobra.Command {
 				node.WithACPType(node.LocalACPType),
 				node.WithPeers(peers...),
 				// db options
-				db.WithUpdateEvents(),
-				db.WithDAGMergeEvents(),
 				db.WithMaxRetries(cfg.GetInt("datastore.MaxTxnRetries")),
 				// net node options
 				net.WithListenAddresses(cfg.GetStringSlice("net.p2pAddresses")...),

client/db.go

@@ -18,7 +18,7 @@ import (
 	"github.com/sourcenetwork/immutable"
 
 	"github.com/sourcenetwork/defradb/datastore"
-	"github.com/sourcenetwork/defradb/events"
+	"github.com/sourcenetwork/defradb/event"
 )
 
 type CollectionName = string
@@ -75,7 +75,7 @@ type DB interface {
 	//
 	// It may be used to monitor database events - a new event will be yielded for each mutation.
 	// Note: it does not copy the queue, just the reference to it.
-	Events() events.Events
+	Events() event.Bus
 
 	// MaxTxnRetries returns the number of retries that this DefraDB instance has been configured to
 	// make in the event of a transaction conflict in certain scenarios.

datastore/mocks/utils.go

@@ -68,6 +68,7 @@ func prepareDAGStore(t *testing.T) *DAGStore {
 func NewTxnWithMultistore(t *testing.T) *MultiStoreTxn {
 	txn := NewTxn(t)
 	txn.EXPECT().OnSuccess(mock.Anything).Maybe()
+	txn.EXPECT().OnSuccessAsync(mock.Anything).Maybe()
 
 	result := &MultiStoreTxn{
 		Txn:             txn,

event/buffered_bus.go

@@ -0,0 +1,141 @@
+// Copyright 2024 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package event
+
+import (
+	"sync/atomic"
+)
+
+type subscribeCommand *Subscription
+
+type unsubscribeCommand *Subscription
+
+type publishCommand Message
+
+type closeCommand struct{}
+
+// bufferedBus is a bus that uses a buffered channel to manage subscribers and publish messages.
+type bufferedBus struct {
+	// subID is incremented for each subscriber and used to set subscriber ids.
+	subID atomic.Uint64
+	// subs is a mapping of subscriber ids to subscriptions.
+	subs map[uint64]*Subscription
+	// events is a mapping of event names to subscriber ids.
+	events map[string]map[uint64]struct{}
+	// commandChannel manages all commands sent to this simpleChannel.
+	//
+	// It is important that all stuff gets sent through this single channel to ensure
+	// that the order of operations is preserved.
+	//
+	// WARNING: This does mean that non-event commands can block the database if the buffer
+	// size is breached (e.g. if many subscribe commands occupy the buffer).
+	commandChannel  chan any
+	eventBufferSize int
+	hasClosedChan   chan struct{}
+	isClosed        atomic.Bool
+}
+
+// NewBufferedBus creates a new event bus with the given commandBufferSize and
+// eventBufferSize.
+//
+// Should the buffers be filled, subsequent calls on this bus will block.
+func NewBufferedBus(commandBufferSize int, eventBufferSize int) Bus {
+	bus := bufferedBus{
+		subs:            make(map[uint64]*Subscription),
+		events:          make(map[string]map[uint64]struct{}),
+		commandChannel:  make(chan any, commandBufferSize),
+		hasClosedChan:   make(chan struct{}),
+		eventBufferSize: eventBufferSize,
+	}
+	go bus.handleChannel()
+	return &bus
+}
+
+func (b *bufferedBus) Publish(msg Message) {
+	if b.isClosed.Load() {
+		return
+	}
+	b.commandChannel <- publishCommand(msg)
+}
+
+func (b *bufferedBus) Subscribe(events ...string) (*Subscription, error) {
+	if b.isClosed.Load() {
+		return nil, ErrSubscribedToClosedChan
+	}
+	sub := &Subscription{
+		id:     b.subID.Add(1),
+		value:  make(chan Message, b.eventBufferSize),
+		events: events,
+	}
+	b.commandChannel <- subscribeCommand(sub)
+	return sub, nil
+}
+
+func (b *bufferedBus) Unsubscribe(sub *Subscription) {
+	if b.isClosed.Load() {
+		return
+	}
+	b.commandChannel <- unsubscribeCommand(sub)
+}
+
+func (b *bufferedBus) Close() {
+	if b.isClosed.Load() {
+		return
+	}
+	b.isClosed.Store(true)
+	b.commandChannel <- closeCommand{}
+	// Wait for the close command to be handled, in order, before returning
+	<-b.hasClosedChan
+}
+
+func (b *bufferedBus) handleChannel() {
+	for cmd := range b.commandChannel {
+		switch t := cmd.(type) {
+		case closeCommand:
+			for _, subscriber := range b.subs {
+				close(subscriber.value)
+			}
+			close(b.commandChannel)
+			close(b.hasClosedChan)
+			return
+
+		case subscribeCommand:
+			for _, event := range t.events {
+				if _, ok := b.events[event]; !ok {
+					b.events[event] = make(map[uint64]struct{})
+				}
+				b.events[event][t.id] = struct{}{}
+			}
+			b.subs[t.id] = t
+
+		case unsubscribeCommand:
+			if _, ok := b.subs[t.id]; !ok {
+				continue // not subscribed
+			}
+			for _, event := range t.events {
+				delete(b.events[event], t.id)
+			}
+			delete(b.subs, t.id)
+			close(t.value)
+
+		case publishCommand:
+			for id := range b.events[WildCardEventName] {
+				b.subs[id].value <- Message(t)
+			}
+			for id := range b.events[t.Name] {
+				if _, ok := b.events[WildCardEventName][id]; ok {
+					continue
+				}
+				b.subs[id].value <- Message(t)
+			}
+		}
+	}
+}

event/buffered_bus_test.go

@@ -0,0 +1,147 @@
+// Copyright 2024 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package event
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSimplePushIsNotBlockedWithoutSubscribers(t *testing.T) {
+	bus := NewBufferedBus(0, 0)
+	defer bus.Close()
+
+	msg := NewMessage("test", 1)
+	bus.Publish(msg)
+
+	// just assert that we reach this line, for the sake of having an assert
+	assert.True(t, true)
+}
+
+func TestSimpleSubscribersAreNotBlockedAfterClose(t *testing.T) {
+	bus := NewBufferedBus(0, 0)
+	defer bus.Close()
+
+	sub, err := bus.Subscribe("test")
+	assert.Nil(t, err)
+
+	bus.Close()
+
+	<-sub.Message()
+
+	// just assert that we reach this line, for the sake of having an assert
+	assert.True(t, true)
+}
+
+func TestSimpleEachSubscribersRecievesEachItem(t *testing.T) {
+	bus := NewBufferedBus(0, 0)
+	defer bus.Close()
+
+	msg1 := NewMessage("test", 1)
+	msg2 := NewMessage("test", 2)
+
+	sub1, err := bus.Subscribe("test")
+	assert.Nil(t, err)
+
+	sub2, err := bus.Subscribe("test")
+	assert.Nil(t, err)
+
+	// ordering of publish is not deterministic
+	// so capture each in a go routine
+	var wg sync.WaitGroup
+	var event1 Message
+	var event2 Message
+
+	go func() {
+		event1 = <-sub1.Message()
+		wg.Done()
+	}()
+
+	go func() {
+		event2 = <-sub2.Message()
+		wg.Done()
+	}()
+
+	wg.Add(2)
+	bus.Publish(msg1)
+	wg.Wait()
+
+	assert.Equal(t, msg1, event1)
+	assert.Equal(t, msg1, event2)
+
+	go func() {
+		event1 = <-sub1.Message()
+		wg.Done()
+	}()
+
+	go func() {
+		event2 = <-sub2.Message()
+		wg.Done()
+	}()
+
+	wg.Add(2)
+	bus.Publish(msg2)
+	wg.Wait()
+
+	assert.Equal(t, msg2, event1)
+	assert.Equal(t, msg2, event2)
+}
+
+func TestSimpleEachSubscribersRecievesEachItemGivenBufferedEventChan(t *testing.T) {
+	bus := NewBufferedBus(0, 2)
+	defer bus.Close()
+
+	msg1 := NewMessage("test", 1)
+	msg2 := NewMessage("test", 2)
+
+	sub1, err := bus.Subscribe("test")
+	assert.Nil(t, err)
+	sub2, err := bus.Subscribe("test")
+	assert.Nil(t, err)
+
+	// both inputs are added first before read, using the internal chan buffer
+	bus.Publish(msg1)
+	bus.Publish(msg2)
+
+	output1Ch1 := <-sub1.Message()
+	output1Ch2 := <-sub2.Message()
+
+	output2Ch1 := <-sub1.Message()
+	output2Ch2 := <-sub2.Message()
+
+	assert.Equal(t, msg1, output1Ch1)
+	assert.Equal(t, msg1, output1Ch2)
+
+	assert.Equal(t, msg2, output2Ch1)
+	assert.Equal(t, msg2, output2Ch2)
+}
+
+func TestSimpleSubscribersDontRecieveItemsAfterUnsubscribing(t *testing.T) {
+	bus := NewBufferedBus(0, 0)
+	defer bus.Close()
+
+	sub, err := bus.Subscribe("test")
+	assert.Nil(t, err)
+	bus.Unsubscribe(sub)
+
+	msg := NewMessage("test", 1)
+	bus.Publish(msg)
+
+	// tiny delay to try and make sure the internal logic would have had time
+	// to do its thing with the pushed item.
+	time.Sleep(5 * time.Millisecond)
+
+	// closing the channel will result in reads yielding the default value
+	assert.Equal(t, Message{}, <-sub.Message())
+}