Update to return an error in the channel

Signed-off-by: Enrique Lacal <enrique.lacal@kaleido.io>

Author: EnriqueL8

internal/blockchain/common/common.go

@@ -44,7 +44,7 @@ type BlockchainCallbacks interface {
 	SetOperationalHandler(namespace string, handler core.OperationCallbacks)
 
 	// BulkOperationUpdates is a way to update multiple operations and get notified when the updates have been committed to the database
-	BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate, onCommit chan<- bool)
+	BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate, onCommit chan<- error)
 
 	OperationUpdate(ctx context.Context, plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject)
 	// Common logic for parsing a BatchPinOrNetworkAction event, and if not discarded to add it to the by-namespace map
@@ -68,7 +68,7 @@ type callbacks struct {
 }
 
 // BulkOperationUpdates implements BlockchainCallbacks.
-func (cb *callbacks) BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate, onCommit chan<- bool) {
+func (cb *callbacks) BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate, onCommit chan<- error) {
 	if handler, ok := cb.opHandlers[namespace]; ok {
 		handler.BulkOperationUpdates(ctx, updates, onCommit)
 		return

internal/operations/manager.go

@@ -47,7 +47,7 @@ type Manager interface {
 	ResubmitOperations(ctx context.Context, txID *fftypes.UUID) (total int, resubmit []*core.Operation, err error)
 	AddOrReuseOperation(ctx context.Context, op *core.Operation, hooks ...database.PostCompletionHook) error
 	BulkInsertOperations(ctx context.Context, ops ...*core.Operation) error
-	SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- bool)
+	SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- error)
 	SubmitOperationUpdate(update *core.OperationUpdate)
 	GetOperationByIDCached(ctx context.Context, opID *fftypes.UUID) (*core.Operation, error)
 	ResolveOperationByID(ctx context.Context, opID *fftypes.UUID, op *core.OperationUpdateDTO) error
@@ -79,7 +79,7 @@ type operationsManager struct {
 }
 
 // SubmitBulkOperationUpdate implements Manager.
-func (om *operationsManager) SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- bool) {
+func (om *operationsManager) SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- error) {
 	for _, update := range updates {
 		errString := ""
 		if update.ErrorMessage != "" {

internal/operations/manager_test.go

@@ -765,7 +765,7 @@ func TestSubmitBulkOperationUpdates(t *testing.T) {
 	}
 
 	// Create a channel to receive the onCommit signal
-	onCommit := make(chan bool, 1)
+	onCommit := make(chan error, 1)
 	go om.SubmitBulkOperationUpdates(ctx, []*core.OperationUpdate{submittedUpdate, submittedUpdate2}, onCommit)
 
 	update := <-om.updater.workQueues[0]
@@ -779,5 +779,6 @@ func TestSubmitBulkOperationUpdates(t *testing.T) {
 	update2.OnComplete()
 
 	// Wait for oncommit signal
-	<-onCommit
+	err := <-onCommit
+	assert.NoError(t, err)
 }

internal/operations/operation_updater.go

@@ -94,7 +94,7 @@ func (ou *operationUpdater) pickWorker(ctx context.Context, id *fftypes.UUID, up
 }
 
 // SubmitBulkOperationUpdates will wait for the commit to DB before calling the onCommit
-func (ou *operationUpdater) SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- bool) {
+func (ou *operationUpdater) SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- error) {
 	validUpdates := []*core.OperationUpdate{}
 	for _, update := range updates {
 		ns, _, err := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
@@ -113,19 +113,24 @@ func (ou *operationUpdater) SubmitBulkOperationUpdates(ctx context.Context, upda
 
 	// Notice how this is not using the workers
 	// The reason is because we want for all updates to be stored at once in this order
-	// If offloaded into worker the updates would be processed in parallel, in different DB TX and in a different order
+	// If offloaded into workers the updates would be processed in parallel, in different DB TX and in a different order
 	go func() {
 		// Copy of the array
+		// TODO make this a deep copy
 		updates := validUpdates
 		// This retries forever until there is no error
 		// but returns on cancelled context
 		err := ou.doBatchUpdateWithRetry(ctx, updates)
 		if err != nil {
 			log.L(ctx).Warnf("Exiting while updating operation: %s", err)
+			if onCommit != nil {
+				onCommit <- err
+			}
+			return
 		}
 		// Batch has been updated correctly
 		if onCommit != nil {
-			onCommit <- true
+			onCommit <- nil
 		}
 	}()
 }
@@ -236,7 +241,6 @@ func (ou *operationUpdater) doBatchUpdateWithRetry(ctx context.Context, updates
 }
 
 func (ou *operationUpdater) doBatchUpdate(ctx context.Context, updates []*core.OperationUpdate) error {
-
 	// Get all the operations that match
 	opIDs := make([]*fftypes.UUID, 0, len(updates))
 	for _, update := range updates {

internal/orchestrator/bound_callbacks.go

@@ -41,7 +41,7 @@ func (bc *boundCallbacks) checkStopped() error {
 	return nil
 }
 
-func (bc *boundCallbacks) BulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- bool) {
+func (bc *boundCallbacks) BulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- error) {
 	bc.o.operations.SubmitBulkOperationUpdates(ctx, updates, onCommit)
 }
 

mocks/coremocks/operation_callbacks.go

@@ -245,7 +245,7 @@ func ParseNamespacedOpID(ctx context.Context, nsIDStr string) (string, *fftypes.
 
 type OperationCallbacks interface {
 	OperationUpdate(update *OperationUpdate)
-	BulkOperationUpdates(ctx context.Context, updates []*OperationUpdate, onCommit chan<- bool)
+	BulkOperationUpdates(ctx context.Context, updates []*OperationUpdate, onCommit chan<- error)
 }
 
 // OperationUpdate notifies FireFly of an update to an operation.