Operation handling for reliable receipts

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/golang-migrate/migrate/v4 v4.17.0
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.1
-	github.com/hyperledger/firefly-common v1.4.14
+	github.com/hyperledger/firefly-common v1.4.15
 	github.com/hyperledger/firefly-signer v1.1.19
 	github.com/jarcoal/httpmock v1.2.0
 	github.com/lib/pq v1.10.9
@@ -72,7 +72,7 @@ require (
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
-	github.com/rs/cors v1.11.0 // indirect
+	github.com/rs/cors v1.11.1 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
@@ -90,7 +90,7 @@ require (
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/term v0.27.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
-	google.golang.org/protobuf v1.32.0 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect

go.sum

@@ -77,8 +77,8 @@ github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+l
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hyperledger/firefly-common v1.4.14 h1:G1x7jKBM2MmbGAo+Hwu/9w3F4cyGuWvYViEZGPLWlic=
-github.com/hyperledger/firefly-common v1.4.14/go.mod h1:tYTzTbVODv/gx0TJ3TkEb+gUieQiAbqLfj/yFNrlDV4=
+github.com/hyperledger/firefly-common v1.4.15 h1:dp4Mo2JQRPMbL7hoMw8T/ktvIUgematOLkXIppQtBp0=
+github.com/hyperledger/firefly-common v1.4.15/go.mod h1:bA7tAJxcpfQMrHN3/YycTSpyk4g2WlnDlpHx8WOUtAY=
 github.com/hyperledger/firefly-signer v1.1.19 h1:Gq5HqUp9/7egLrahJY9WMk4Y9dZVPIl99aSIged93HM=
 github.com/hyperledger/firefly-signer v1.1.19/go.mod h1:XTwaPRkAfVxk2G3PQOYHLbuvMOiBs0px/4vwXTsUtsA=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
@@ -163,8 +163,8 @@ github.com/qeesung/image2ascii v1.0.1 h1:Fe5zTnX/v/qNC3OC4P/cfASOXS501Xyw2UUcgrL
 github.com/qeesung/image2ascii v1.0.1/go.mod h1:kZKhyX0h2g/YXa/zdJR3JnLnJ8avHjZ3LrvEKSYyAyU=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
-github.com/rs/cors v1.11.0 h1:0B9GE/r9Bc2UxRMMtymBkHTenPkHDv0CW4Y98GBY+po=
-github.com/rs/cors v1.11.0/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
+github.com/rs/cors v1.11.1 h1:eU3gRzXLRK57F5rKMGMZURNdIG4EoAmX8k94r9wXWHA=
+github.com/rs/cors v1.11.1/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sagikazarmark/locafero v0.4.0 h1:HApY1R9zGo4DBgr7dqsTH/JJxLTTsOt7u6keLGt6kNQ=
 github.com/sagikazarmark/locafero v0.4.0/go.mod h1:Pe1W6UlPYUk/+wc/6KFhbORCfqzgYEpgQ3O5fPuL3H4=
@@ -278,8 +278,8 @@ golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
-google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

go.work.sum

@@ -253,10 +253,8 @@ github.com/hashicorp/go-memdb v1.3.3/go.mod h1:uBTr1oQbtuMgd1SSGoR8YV27eT3sBHbYi
 github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/serf v0.10.1/go.mod h1:yL2t6BqATOLGc5HF7qbFkTfXoPIY0WZdWHfEvMqbG+4=
-github.com/hyperledger/firefly-common v1.4.14 h1:G1x7jKBM2MmbGAo+Hwu/9w3F4cyGuWvYViEZGPLWlic=
-github.com/hyperledger/firefly-common v1.4.14/go.mod h1:tYTzTbVODv/gx0TJ3TkEb+gUieQiAbqLfj/yFNrlDV4=
-github.com/hyperledger/firefly-signer v1.1.19 h1:Gq5HqUp9/7egLrahJY9WMk4Y9dZVPIl99aSIged93HM=
-github.com/hyperledger/firefly-signer v1.1.19/go.mod h1:XTwaPRkAfVxk2G3PQOYHLbuvMOiBs0px/4vwXTsUtsA=
+github.com/hyperledger/firefly-common v1.4.15 h1:dp4Mo2JQRPMbL7hoMw8T/ktvIUgematOLkXIppQtBp0=
+github.com/hyperledger/firefly-common v1.4.15/go.mod h1:bA7tAJxcpfQMrHN3/YycTSpyk4g2WlnDlpHx8WOUtAY=
 github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
 github.com/jackc/pgconn v1.14.0/go.mod h1:9mBNlny0UvkgJdCDvdVHYSjI+8tD2rnKK69Wz8ti++E=
 github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
@@ -303,6 +301,8 @@ github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8/go.mod h1:HKlIX3XHQyzL
 github.com/pkg/sftp v1.13.6/go.mod h1:tz1ryNURKu77RL+GuCzmoJYxQczL3wLNNpPWagdg4Qk=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rqlite/gorqlite v0.0.0-20230708021416-2acd02b70b79/go.mod h1:xF/KoXmrRyahPfo5L7Szb5cAAUl53dMWBh9cMruGEZg=
+github.com/rs/cors v1.11.1 h1:eU3gRzXLRK57F5rKMGMZURNdIG4EoAmX8k94r9wXWHA=
+github.com/rs/cors v1.11.1/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/sagikazarmark/crypt v0.17.0/go.mod h1:SMtHTvdmsZMuY/bpZoqokSoChIrcJ/epOxZN58PbZDg=
 github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/snowflakedb/gosnowflake v1.6.19/go.mod h1:FM1+PWUdwB9udFDsXdfD58NONC0m+MlOSmQRvimobSM=
@@ -335,6 +335,8 @@ google.golang.org/appengine v1.6.8/go.mod h1:1jJ3jBArFh5pcgW8gCtRJnepW8FzD1V44FJ
 google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac h1:ZL/Teoy/ZGnzyrqK/Optxxp2pmVh+fmJ97slxSRyzUg=
 google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac/go.mod h1:+Rvu7ElI+aLzyDQhpHMFMMltsD6m7nqpuWDd2CwJw3k=
 google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:0xJLfVdJqpAPl8tDg1ujOCGzx6LFLttXT5NhllGOXY4=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/bson.v2 v2.0.0-20171018101713-d8c8987b8862/go.mod h1:VN8wuk/3Ksp8lVZ82HHf/MI1FHOBDt5bPK9VZ8DvymM=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=

internal/blockchain/common/common.go

@@ -1,4 +1,4 @@
-// Copyright © 2024 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -43,6 +43,9 @@ type BlockchainCallbacks interface {
 	SetHandler(namespace string, handler blockchain.Callbacks)
 	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)
+
 	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
 	PrepareBatchPinOrNetworkAction(ctx context.Context, events EventsToDispatch, subInfo *SubscriptionInfo, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef, params *BatchPinParams)
@@ -64,6 +67,16 @@ type callbacks struct {
 	opHandlers map[string]core.OperationCallbacks
 }
 
+// BulkOperationUpdates implements BlockchainCallbacks.
+func (cb *callbacks) BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate, onCommit chan<- bool) {
+	if handler, ok := cb.opHandlers[namespace]; ok {
+		handler.BulkOperationUpdates(ctx, updates, onCommit)
+		return
+	}
+	// Need to clean this up
+	log.L(ctx).Errorf("No handler found for namespace '%s'", namespace)
+}
+
 type subscriptions struct {
 	subs map[string]*SubscriptionInfo
 }

internal/operations/manager.go

@@ -1,4 +1,4 @@
 // Copyright © 2024 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -47,6 +47,7 @@
 	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)
 	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
@@ -77,6 +78,18 @@
 	cache     cache.CInterface
 }
 
+// SubmitBulkOperationUpdate implements Manager.
+func (om *operationsManager) SubmitBulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- bool) {
+	for _, update := range updates {
+		errString := ""
+		if update.ErrorMessage != "" {
+			errString = fmt.Sprintf(" error=%s", update.ErrorMessage)
+		}
+		log.L(om.ctx).Debugf("%s updating operation %s status=%s%s", update.Plugin, update.NamespacedOpID, update.Status, errString)
+	}
+	om.updater.SubmitBulkOperationUpdates(ctx, updates, onCommit)
+}
+
 func NewOperationsManager(ctx context.Context, ns string, di database.Plugin, txHelper txcommon.Helper, cacheManager cache.Manager) (Manager, error) {
 	if di == nil || txHelper == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "OperationsManager")

internal/operations/manager_test.go

@@ -742,3 +742,42 @@ func TestErrTernaryHelper(t *testing.T) {
 	assert.Equal(t, core.OpPhasePending, ErrTernary(nil, core.OpPhaseInitializing, core.OpPhasePending))
 	assert.Equal(t, core.OpPhaseInitializing, ErrTernary(fmt.Errorf("pop"), core.OpPhaseInitializing, core.OpPhasePending))
 }
+
+func TestSubmitBulkOperationUpdates(t *testing.T) {
+	om, cancel := newTestOperations(t)
+	defer cancel()
+
+	om.updater.workQueues = []chan *core.OperationUpdate{
+		make(chan *core.OperationUpdate, 1),
+	}
+
+	ctx := context.Background()
+	submittedUpdate := &core.OperationUpdate{
+		NamespacedOpID: "ns1:" + fftypes.NewUUID().String(),
+		Status:         core.OpStatusSucceeded,
+		ErrorMessage:   "my-error-message",
+	}
+
+	submittedUpdate2 := &core.OperationUpdate{
+		NamespacedOpID: "ns1:" + fftypes.NewUUID().String(),
+		Status:         core.OpStatusSucceeded,
+		ErrorMessage:   "my-error-message",
+	}
+
+	// Create a channel to receive the onCommit signal
+	onCommit := make(chan bool, 1)
+	go om.SubmitBulkOperationUpdates(ctx, []*core.OperationUpdate{submittedUpdate, submittedUpdate2}, onCommit)
+
+	update := <-om.updater.workQueues[0]
+	assert.Equal(t, submittedUpdate.NamespacedOpID, update.NamespacedOpID)
+	assert.Equal(t, core.OpStatusSucceeded, update.Status)
+	update.OnComplete()
+
+	update2 := <-om.updater.workQueues[0]
+	assert.Equal(t, submittedUpdate2.NamespacedOpID, update2.NamespacedOpID)
+	assert.Equal(t, core.OpStatusSucceeded, update2.Status)
+	update2.OnComplete()
+
+	// Wait for oncommit signal
+	<-onCommit
+}

internal/operations/operation_updater.go

@@ -1,4 +1,4 @@
 // Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -93,6 +93,43 @@
 	return ou.workQueues[worker]
 }
 
+// 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) {
+	validUpdates := []*core.OperationUpdate{}
+	for _, update := range updates {
+		ns, _, err := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
+		if err != nil {
+			log.L(ctx).Warnf("Unable to update operation '%s' due to invalid ID: %s", update.NamespacedOpID, err)
+			continue
+		}
+
+		if ns != ou.manager.namespace {
+			log.L(ou.ctx).Debugf("Ignoring operation update from different namespace '%s'", ns)
+			continue
+		}
+
+		validUpdates = append(validUpdates, update)
+	}
+
+	// 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
+	go func() {
+		// Copy of the array
+		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)
+		}
+		// Batch has been updated correctly
+		if onCommit != nil {
+			onCommit <- true
+		}
+	}()
+}
+
 func (ou *operationUpdater) SubmitOperationUpdate(ctx context.Context, update *core.OperationUpdate) {
 	ns, id, err := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
 	if err != nil {

internal/orchestrator/bound_callbacks.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -41,6 +41,10 @@ func (bc *boundCallbacks) checkStopped() error {
 	return nil
 }
 
+func (bc *boundCallbacks) BulkOperationUpdates(ctx context.Context, updates []*core.OperationUpdate, onCommit chan<- bool) {
+	bc.o.operations.SubmitBulkOperationUpdates(ctx, updates, onCommit)
+}
+
 func (bc *boundCallbacks) OperationUpdate(update *core.OperationUpdate) {
 	bc.o.operations.SubmitOperationUpdate(update)
 }