feat: batching events and webhook plugin support

Signed-off-by: Enrique Lacal <[email protected]>

Author: EnriqueL8

docs/reference/types/subscription.md

@@ -105,6 +105,8 @@ nav_order: 3
 | `firstEvent` | Whether your application would like to receive events from the 'oldest' event emitted by your FireFly node (from the beginning of time), or the 'newest' event (from now), or a specific event sequence. Default is 'newest' | `SubOptsFirstEvent` |
 | `readAhead` | The number of events to stream ahead to your application, while waiting for confirmation of consumption of those events. At least once delivery semantics are used in FireFly, so if your application crashes/reconnects this is the maximum number of events you would expect to be redelivered after it restarts | `uint16` |
 | `withData` | Whether message events delivered over the subscription, should be packaged with the full data of those messages in-line as part of the event JSON payload. Or if the application should make separate REST calls to download that data. May not be supported on some transports. | `bool` |
+| `batch` | Enable batching. Works in conjunction with readAhead which defines the batchSize. | `bool` |
+| `batchTimeout` | When batching is enabled, the optional timeout to send events even when the batch hasn't filled. | `string` |
 | `fastack` | Webhooks only: When true the event will be acknowledged before the webhook is invoked, allowing parallel invocations | `bool` |
 | `url` | Webhooks only: HTTP url to invoke. Can be relative if a base URL is set in the webhook plugin config | `string` |
 | `method` | Webhooks only: HTTP method to invoke. Default=POST | `string` |
@@ -148,7 +150,7 @@ nav_order: 3
 | `requestTimeout` | The max duration to hold a TLS handshake alive | `string` |
 | `maxIdleConns` | The max number of idle connections to hold pooled | `int` |
 | `idleTimeout` | The max duration to hold a HTTP keepalive connection between calls | `string` |
-| `connectionTimeout` |  | `string` |
+| `connectionTimeout` | The maximum amount of time that a connection is allowed to remain with no data transmitted. | `string` |
 | `expectContinueTimeout` | See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport) | `string` |
 
 

docs/reference/types/wsstart.md

@@ -96,6 +96,8 @@ nav_order: 23
 | `firstEvent` | Whether your application would like to receive events from the 'oldest' event emitted by your FireFly node (from the beginning of time), or the 'newest' event (from now), or a specific event sequence. Default is 'newest' | `SubOptsFirstEvent` |
 | `readAhead` | The number of events to stream ahead to your application, while waiting for confirmation of consumption of those events. At least once delivery semantics are used in FireFly, so if your application crashes/reconnects this is the maximum number of events you would expect to be redelivered after it restarts | `uint16` |
 | `withData` | Whether message events delivered over the subscription, should be packaged with the full data of those messages in-line as part of the event JSON payload. Or if the application should make separate REST calls to download that data. May not be supported on some transports. | `bool` |
+| `batch` | Enable batching. Works in conjunction with readAhead which defines the batchSize. | `bool` |
+| `batchTimeout` | When batching is enabled, the optional timeout to send events even when the batch hasn't filled. | `string` |
 | `fastack` | Webhooks only: When true the event will be acknowledged before the webhook is invoked, allowing parallel invocations | `bool` |
 | `url` | Webhooks only: HTTP url to invoke. Can be relative if a base URL is set in the webhook plugin config | `string` |
 | `method` | Webhooks only: HTTP method to invoke. Default=POST | `string` |
@@ -139,7 +141,7 @@ nav_order: 23
 | `requestTimeout` | The max duration to hold a TLS handshake alive | `string` |
 | `maxIdleConns` | The max number of idle connections to hold pooled | `int` |
 | `idleTimeout` | The max duration to hold a HTTP keepalive connection between calls | `string` |
-| `connectionTimeout` |  | `string` |
+| `connectionTimeout` | The maximum amount of time that a connection is allowed to remain with no data transmitted. | `string` |
 | `expectContinueTimeout` | See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport) | `string` |
 
 

docs/swagger/swagger.yaml

@@ -537,9 +537,11 @@ var (
 	SubscriptionBlockchainEventFilterListener = ffm("SubscriptionBlockchainEventFilter.listener", "Regular expression to apply to the blockchain event 'listener' field, which is the UUID of the event listener. So you can restrict your subscription to certain blockchain listeners. Alternatively to avoid your application need to know listener UUIDs you can set the 'topic' field of blockchain event listeners, and use a topic filter on your subscriptions")
 
 	// SubscriptionCoreOptions field descriptions
-	SubscriptionCoreOptionsFirstEvent = ffm("SubscriptionCoreOptions.firstEvent", "Whether your application would like to receive events from the 'oldest' event emitted by your FireFly node (from the beginning of time), or the 'newest' event (from now), or a specific event sequence. Default is 'newest'")
-	SubscriptionCoreOptionsReadAhead  = ffm("SubscriptionCoreOptions.readAhead", "The number of events to stream ahead to your application, while waiting for confirmation of consumption of those events. At least once delivery semantics are used in FireFly, so if your application crashes/reconnects this is the maximum number of events you would expect to be redelivered after it restarts")
-	SubscriptionCoreOptionsWithData   = ffm("SubscriptionCoreOptions.withData", "Whether message events delivered over the subscription, should be packaged with the full data of those messages in-line as part of the event JSON payload. Or if the application should make separate REST calls to download that data. May not be supported on some transports.")
+	SubscriptionCoreOptionsFirstEvent   = ffm("SubscriptionCoreOptions.firstEvent", "Whether your application would like to receive events from the 'oldest' event emitted by your FireFly node (from the beginning of time), or the 'newest' event (from now), or a specific event sequence. Default is 'newest'")
+	SubscriptionCoreOptionsReadAhead    = ffm("SubscriptionCoreOptions.readAhead", "The number of events to stream ahead to your application, while waiting for confirmation of consumption of those events. At least once delivery semantics are used in FireFly, so if your application crashes/reconnects this is the maximum number of events you would expect to be redelivered after it restarts")
+	SubscriptionCoreOptionsWithData     = ffm("SubscriptionCoreOptions.withData", "Whether message events delivered over the subscription, should be packaged with the full data of those messages in-line as part of the event JSON payload. Or if the application should make separate REST calls to download that data. May not be supported on some transports.")
+	SubscriptionCoreOptionsBatch        = ffm("SubscriptionCoreOptions.batch", "Enable batching. Works in conjunction with readAhead which defines the batchSize.")
+	SubscriptionCoreOptionsBatchTimeout = ffm("SubscriptionCoreOptions.batchTimeout", "When batching is enabled, the optional timeout to send events even when the batch hasn't filled.")
 
 	// TokenApproval field descriptions
 	TokenApprovalLocalID         = ffm("TokenApproval.localId", "The UUID of this token approval, in the local FireFly node")
@@ -706,7 +708,7 @@ var (
 	WebhookOptHTTPExpectContinueTimeout = ffm("WebhookHTTPOptions.expectContinueTimeout", "See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport)")
 	WebhookOptHTTPIdleTimeout           = ffm("WebhookHTTPOptions.idleTimeout", "The max duration to hold a HTTP keepalive connection between calls")
 	WebhookOptHTTPMaxIdleConns          = ffm("WebhookHTTPOptions.maxIdleConns", "The max number of idle connections to hold pooled")
-	WebhookOptHTTPConnectionTimeout     = ffm("WebhookHTTPOptions.connectionTimeout", "")
+	WebhookOptHTTPConnectionTimeout     = ffm("WebhookHTTPOptions.connectionTimeout", "The maximum amount of time that a connection is allowed to remain with no data transmitted.")
 	WebhookOptHTTPTLSHandshakeTimeout   = ffm("WebhookHTTPOptions.tlsHandshakeTimeout", "The max duration to hold a TLS handshake alive")
 	WebhookOptHTTPRequestTimeout        = ffm("WebhookHTTPOptions.requestTimeout", "The max duration to hold a TLS handshake alive")
 

internal/coremsgs/en_struct_descriptions.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
@@ -39,7 +40,8 @@ import (
 )
 
 const (
-	maxReadAhead = 65536
+	maxReadAhead        = 65536
+	defaultBatchTimeout = time.Duration(2) * time.Second
 )
 
 type ackNack struct {
@@ -66,7 +68,9 @@ type eventDispatcher struct {
 	eventDelivery chan *core.EventDelivery
 	mux           sync.Mutex
 	namespace     string
-	readAhead     int
+	readAhead     int  // confusing to have both
+	batch         bool // boolean? gonna receiver array, interface of application to FF
+	batchTimeout  time.Duration
 	subscription  *subscription
 	txHelper      txcommon.Helper
 }
@@ -80,6 +84,11 @@ func newEventDispatcher(ctx context.Context, enricher *eventEnricher, ei events.
 	if readAhead > maxReadAhead {
 		readAhead = maxReadAhead
 	}
+
+	batchTimeout := defaultBatchTimeout
+	if sub.definition.Options.BatchTimeout != "" {
+		batchTimeout = fftypes.ParseToDuration(sub.definition.Options.BatchTimeout)
+	}
 	ed := &eventDispatcher{
 		ctx: log.WithLogField(log.WithLogField(ctx,
 			"role", fmt.Sprintf("ed[%s]", connID)),
@@ -100,6 +109,8 @@ func newEventDispatcher(ctx context.Context, enricher *eventEnricher, ei events.
 		acksNacks:     make(chan ackNack),
 		closed:        make(chan struct{}),
 		txHelper:      txHelper,
+		batch:         sub.definition.Options.Batch,
+		batchTimeout:  batchTimeout,
 	}
 
 	pollerConf := &eventPollerConf{
@@ -149,7 +160,12 @@ func (ed *eventDispatcher) electAndStart() {
 	// We're ready to go - not
 	ed.elected = true
 	ed.eventPoller.start()
-	go ed.deliverEvents()
+
+	if ed.batch {
+		go ed.deliverBatchedEvents()
+	} else {
+		go ed.deliverEvents()
+	}
 	// Wait until the event poller closes
 	<-ed.eventPoller.closed
 }
@@ -262,6 +278,7 @@ func (ed *eventDispatcher) bufferedDelivery(events []core.LocallySequenced) (boo
 	// At this point, the page of messages we've been given are loaded from the DB into memory,
 	// but we can only make them in-flight and push them to the client up to the maximum
 	// readahead (which is likely lower than our page size - 1 by default)
+	// TODO how do we guarantee that the readahead is lower that our page size?
 
 	if len(events) == 0 {
 		return false, nil
@@ -280,33 +297,35 @@ func (ed *eventDispatcher) bufferedDelivery(events []core.LocallySequenced) (boo
 	matchCount := len(matching)
 	dispatched := 0
 
+	// TODO We handle batch delivery here!
+
 	// We stay here blocked until we've consumed all the messages in the buffer,
 	// or a reset event happens
 	for {
 		ed.mux.Lock()
-		var disapatchable []*core.EventDelivery
+		var dispatchable []*core.EventDelivery
 		inflightCount := len(ed.inflight)
 		maxDispatch := 1 + ed.readAhead - inflightCount
 		if maxDispatch >= len(matching) {
-			disapatchable = matching
+			dispatchable = matching
 			matching = nil
 		} else if maxDispatch > 0 {
-			disapatchable = matching[0:maxDispatch]
+			dispatchable = matching[0:maxDispatch]
 			matching = matching[maxDispatch:]
 		}
 		ed.mux.Unlock()
 
 		l.Debugf("Dispatcher event state: readahead=%d candidates=%d matched=%d inflight=%d queued=%d dispatched=%d dispatchable=%d lastAck=%d nacks=%d highest=%d",
-			ed.readAhead, len(candidates), matchCount, inflightCount, len(matching), dispatched, len(disapatchable), lastAck, nacks, highestOffset)
+			ed.readAhead, len(candidates), matchCount, inflightCount, len(matching), dispatched, len(dispatchable), lastAck, nacks, highestOffset)
 
-		for _, event := range disapatchable {
+		for _, event := range dispatchable {
 			ed.mux.Lock()
 			ed.inflight[*event.ID] = &event.Event
 			inflightCount = len(ed.inflight)
 			ed.mux.Unlock()
 
 			dispatched++
-			ed.eventDelivery <- event
+			ed.eventDelivery <- event // Do we send batches on the channel or accumulate in the plugin?
 		}
 
 		if inflightCount == 0 {
@@ -364,6 +383,72 @@ func (ed *eventDispatcher) handleAckOffsetUpdate(ack ackNack) {
 	}
 }
 
+func (ed *eventDispatcher) deliverBatchedEvents() {
+	withData := ed.subscription.definition.Options.WithData != nil && *ed.subscription.definition.Options.WithData
+
+	var events []*core.EventDelivery
+	var dataSet []core.DataArray
+	var batchTimeoutContext context.Context
+	var batchTimeoutCancel func()
+	for {
+		var timeoutContext context.Context
+		var timedOut bool
+		if batchTimeoutContext != nil {
+			timeoutContext = batchTimeoutContext
+		} else {
+			timeoutContext = ed.ctx
+		}
+		select {
+		case event, ok := <-ed.eventDelivery:
+			if !ok {
+				if batchTimeoutCancel != nil {
+					batchTimeoutCancel()
+				}
+				return
+			}
+
+			if events == nil && dataSet == nil {
+				events = []*core.EventDelivery{}
+				dataSet = []core.DataArray{}
+				batchTimeoutContext, batchTimeoutCancel = context.WithTimeout(ed.ctx, ed.batchTimeout)
+			}
+
+			log.L(ed.ctx).Debugf("Dispatching %s event in a batch: %.10d/%s [%s]: ref=%s/%s", ed.transport.Name(), event.Sequence, event.ID, event.Type, event.Namespace, event.Reference)
+
+			events = append(events, event)
+
+			var data []*core.Data
+			var err error
+			if withData && event.Message != nil {
+				data, _, err = ed.data.GetMessageDataCached(ed.ctx, event.Message)
+				dataSet = append(dataSet, data)
+			}
+
+			if err != nil {
+				ed.deliveryResponse(&core.EventDeliveryResponse{ID: event.ID, Rejected: true})
+			}
+
+		case <-timeoutContext.Done():
+			timedOut = true
+		case <-ed.ctx.Done():
+			if batchTimeoutCancel != nil {
+				batchTimeoutCancel()
+			}
+			return
+		}
+
+		if len(events) >= ed.readAhead || (timedOut && len(events) > 0) {
+			// TODO properly handle the error
+			batchTimeoutCancel()
+			_ = ed.transport.DeliveryBatchRequest(ed.ctx, ed.connID, ed.subscription.definition, events, dataSet)
+			// If err handle all the delivery responses for all the events??
+			events = nil
+			dataSet = nil
+		}
+	}
+}
+
+// TODO issue here, we can't just call DeliveryRequest with one thing.
 func (ed *eventDispatcher) deliverEvents() {
 	withData := ed.subscription.definition.Options.WithData != nil && *ed.subscription.definition.Options.WithData
 	for {
@@ -372,12 +457,14 @@ func (ed *eventDispatcher) deliverEvents() {
 			if !ok {
 				return
 			}
+
 			log.L(ed.ctx).Debugf("Dispatching %s event: %.10d/%s [%s]: ref=%s/%s", ed.transport.Name(), event.Sequence, event.ID, event.Type, event.Namespace, event.Reference)
 			var data []*core.Data
 			var err error
 			if withData && event.Message != nil {
 				data, _, err = ed.data.GetMessageDataCached(ed.ctx, event.Message)
 			}
+
 			if err == nil {
 				err = ed.transport.DeliveryRequest(ed.ctx, ed.connID, ed.subscription.definition, event, data)
 			}

internal/events/event_dispatcher.go

@@ -1064,3 +1064,55 @@ func TestEventDispatcherWithReply(t *testing.T) {
 	mbm.AssertExpectations(t)
 	mms.AssertExpectations(t)
 }
+
+func TestEventDeliveryBatch(t *testing.T) {
+	log.SetLevel("debug")
+	var five = uint16(5)
+	sub := &subscription{
+		dispatcherElection: make(chan bool, 1),
+		definition: &core.Subscription{
+			SubscriptionRef: core.SubscriptionRef{ID: fftypes.NewUUID(), Namespace: "ns1", Name: "sub1"},
+			Options: core.SubscriptionOptions{
+				SubscriptionCoreOptions: core.SubscriptionCoreOptions{
+					ReadAhead: &five,
+					Batch:     true,
+				},
+			},
+		},
+		eventMatcher: regexp.MustCompile(fmt.Sprintf("^%s|%s$", core.EventTypeMessageConfirmed, core.EventTypeMessageConfirmed)),
+	}
+
+	ed, cancel := newTestEventDispatcher(sub)
+	cancel()
+	ed.acksNacks = make(chan ackNack, 5)
+
+	event1 := fftypes.NewUUID()
+	ed.inflight[*event1] = &core.Event{
+		ID:        event1,
+		Namespace: "ns1",
+	}
+
+	mms := &syncasyncmocks.Sender{}
+	mbm := ed.broadcast.(*broadcastmocks.Manager)
+	mbm.On("NewBroadcast", mock.Anything).Return(mms)
+	mms.On("Send", mock.Anything).Return(nil)
+
+	ed.deliveryResponse(&core.EventDeliveryResponse{
+		ID: event1,
+		Reply: &core.MessageInOut{
+			Message: core.Message{
+				Header: core.MessageHeader{
+					Tag:  "myreplytag1",
+					CID:  fftypes.NewUUID(),
+					Type: core.MessageTypeBroadcast,
+				},
+			},
+			InlineData: core.InlineData{
+				{Value: fftypes.JSONAnyPtr(`"my reply"`)},
+			},
+		},
+	})
+
+	mbm.AssertExpectations(t)
+	mms.AssertExpectations(t)
+}

internal/events/event_dispatcher_test.go

@@ -134,6 +134,10 @@ func (se *Events) DeliveryRequest(ctx context.Context, connID string, sub *core.
 	return nil
 }
 
+func (se *Events) DeliveryBatchRequest(ctx context.Context, connID string, sub *core.Subscription, events []*core.EventDelivery, data []core.DataArray) error {
+	return nil
+}
+
 func (se *Events) NamespaceRestarted(ns string, startTime time.Time) {
 	// no-op
 }