v1.1: Replace "UpsertBatch" with "InsertOrGetBatch"

internal/batch/batch_manager_test.go

@@ -6,7 +6,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -132,7 +132,7 @@ func TestE2EDispatchBroadcast(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{{ID: *msg.Header.ID}}, nil).Once()
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{}, nil)
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("UpdateBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil) // pins
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything).Return(nil)
@@ -255,7 +255,7 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{{ID: *msg.Header.ID}}, nil).Once()
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{}, nil)
 	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil) // pins
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("UpdateBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	rag.RunFn = func(a mock.Arguments) {
@@ -441,7 +441,7 @@ func TestMessageSequencerUpdateMessagesFail(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 	mdi.On("InsertTransaction", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil) // transaction submit
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("fizzle"))
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything)
 	rag.RunFn = func(a mock.Arguments) {
@@ -538,7 +538,7 @@ func TestMessageSequencerUpdateBatchFail(t *testing.T) {
 	}
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{{ID: *msg.Header.ID}}, nil)
 	mdm.On("GetMessageWithDataCached", mock.Anything, mock.Anything).Return(msg, core.DataArray{{ID: dataID}}, true, nil)
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("fizzle"))
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, fmt.Errorf("fizzle"))
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything)
 	rag.RunFn = func(a mock.Arguments) {
 		ctx := a.Get(0).(context.Context)

internal/batch/batch_processor.go

@@ -576,7 +576,8 @@ func (bp *batchProcessor) sealBatch(state *DispatchState) (err error) {
 			log.L(ctx).Debugf("Batch %s sealed. Hash=%s", state.Persisted.ID, state.Persisted.Hash)
 
 			// At this point the manifest of the batch is finalized. We write it to the database
-			return bp.database.UpsertBatch(ctx, &state.Persisted)
+			_, err = bp.database.InsertOrGetBatch(ctx, &state.Persisted)
+			return err
 		})
 	})
 	if err != nil {

internal/batch/batch_processor_test.go

@@ -6,7 +6,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -85,7 +85,7 @@ func TestUnfilledBatch(t *testing.T) {
 
 	mockRunAsGroupPassthrough(mdi)
 	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 
 	mth := bp.txHelper.(*txcommonmocks.Helper)
 	mth.On("SubmitNewTransaction", mock.Anything, core.TransactionTypeBatchPin, core.IdempotencyKey("")).Return(fftypes.NewUUID(), nil)
@@ -134,7 +134,7 @@ func TestBatchSizeOverflow(t *testing.T) {
 	bp.conf.BatchMaxBytes = batchSizeEstimateBase + (&core.Message{}).EstimateSize(false) + 100
 	mockRunAsGroupPassthrough(mdi)
 	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 
 	mth := bp.txHelper.(*txcommonmocks.Helper)
 	mth.On("SubmitNewTransaction", mock.Anything, core.TransactionTypeBatchPin, core.IdempotencyKey("")).Return(fftypes.NewUUID(), nil)
@@ -419,7 +419,7 @@ func TestMarkMessageDispatchedUnpinnedOK(t *testing.T) {
 
 	mockRunAsGroupPassthrough(mdi)
 	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(fmt.Errorf("pop")).Once()
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
 
@@ -476,7 +476,7 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 		return dbNonce.Nonce == 12347 // twice incremented
 	})).Return(nil).Once()
 	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil).Twice()
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 
 	mdm := bp.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()

internal/database/sqlcommon/batch_sql.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -56,87 +56,49 @@ var (
 
 const batchesTable = "batches"
 
-func (s *SQLCommon) UpsertBatch(ctx context.Context, batch *core.BatchPersisted) (err error) {
+func (s *SQLCommon) InsertOrGetBatch(ctx context.Context, batch *core.BatchPersisted) (existing *core.BatchPersisted, err error) {
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer s.rollbackTx(ctx, tx, autoCommit)
 
-	// Do a select within the transaction to detemine if the UUID already exists
-	batchRows, _, err := s.queryTx(ctx, batchesTable, tx,
-		sq.Select("hash").
-			From(batchesTable).
-			Where(sq.Eq{"id": batch.ID, "namespace": batch.Namespace}),
+	// Try the insert first
+	_, insertErr := s.insertTxExt(ctx, batchesTable, tx,
+		sq.Insert(batchesTable).
+			Columns(batchColumns...).
+			Values(
+				batch.ID,
+				string(batch.Type),
+				batch.Namespace,
+				batch.Author,
+				batch.Key,
+				batch.Group,
+				batch.Created,
+				batch.Hash,
+				batch.Manifest,
+				batch.Confirmed,
+				batch.TX.Type,
+				batch.TX.ID,
+				batch.Node,
+			),
+		func() {
+			s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeCreated, batch.Namespace, batch.ID)
+		},
+		true, /* we want a failure here we can progress past */
 	)
-	if err != nil {
-		return err
+	if insertErr == nil {
+		return nil, s.commitTx(ctx, tx, autoCommit)
 	}
 
-	existing := batchRows.Next()
-	if existing {
-		var hash *fftypes.Bytes32
-		_ = batchRows.Scan(&hash)
-		if !fftypes.SafeHashCompare(hash, batch.Hash) {
-			batchRows.Close()
-			log.L(ctx).Errorf("Existing=%s New=%s", hash, batch.Hash)
-			return database.HashMismatch
-		}
-	}
-	batchRows.Close()
-
-	if existing {
-
-		// Update the batch
-		if _, err = s.updateTx(ctx, batchesTable, tx,
-			sq.Update(batchesTable).
-				Set("btype", string(batch.Type)).
-				Set("author", batch.Author).
-				Set("key", batch.Key).
-				Set("group_hash", batch.Group).
-				Set("created", batch.Created).
-				Set("hash", batch.Hash).
-				Set("manifest", batch.Manifest).
-				Set("confirmed", batch.Confirmed).
-				Set("tx_type", batch.TX.Type).
-				Set("tx_id", batch.TX.ID).
-				Set("node_id", batch.Node).
-				Where(sq.Eq{"id": batch.ID, "namespace": batch.Namespace}),
-			func() {
-				s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeUpdated, batch.Namespace, batch.ID)
-			},
-		); err != nil {
-			return err
-		}
-	} else {
-
-		if _, err = s.insertTx(ctx, batchesTable, tx,
-			sq.Insert(batchesTable).
-				Columns(batchColumns...).
-				Values(
-					batch.ID,
-					string(batch.Type),
-					batch.Namespace,
-					batch.Author,
-					batch.Key,
-					batch.Group,
-					batch.Created,
-					batch.Hash,
-					batch.Manifest,
-					batch.Confirmed,
-					batch.TX.Type,
-					batch.TX.ID,
-					batch.Node,
-				),
-			func() {
-				s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeCreated, batch.Namespace, batch.ID)
-			},
-		); err != nil {
-			return err
-		}
+	// Do a select within the transaction to determine if the batch already exists
+	existing, err = s.GetBatchByID(ctx, batch.Namespace, batch.ID)
+	if err != nil || existing != nil {
+		return existing, err
 	}
 
-	return s.commitTx(ctx, tx, autoCommit)
+	// Error was apparently not an ID conflict - must have been something else
+	return nil, insertErr
 }
 
 func (s *SQLCommon) batchResult(ctx context.Context, row *sql.Rows) (*core.BatchPersisted, error) {

internal/database/sqlcommon/batch_sql_test.go

@@ -63,10 +63,10 @@ func TestBatch2EWithDB(t *testing.T) {
 	}
 
 	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBatches, core.ChangeEventTypeCreated, "ns1", batchID, mock.Anything).Return()
-	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBatches, core.ChangeEventTypeUpdated, "ns1", batchID, mock.Anything).Return()
 
-	err := s.UpsertBatch(ctx, batch)
+	existing, err := s.InsertOrGetBatch(ctx, batch)
 	assert.NoError(t, err)
+	assert.Nil(t, existing)
 
 	// Check we get the exact same batch back
 	batchRead, err := s.GetBatchByID(ctx, "ns1", batchID)
@@ -76,58 +76,17 @@ func TestBatch2EWithDB(t *testing.T) {
 	batchReadJson, _ := json.Marshal(&batchRead)
 	assert.Equal(t, string(batchJson), string(batchReadJson))
 
-	// Update the batch (this is testing what's possible at the database layer,
-	// and does not account for the verification that happens at the higher level)
-	txid := fftypes.NewUUID()
-	msgID2 := fftypes.NewUUID()
-	batchUpdated := &core.BatchPersisted{
-		BatchHeader: core.BatchHeader{
-			ID:   batchID,
-			Type: core.BatchTypePrivate,
-			SignerRef: core.SignerRef{
-				Key:    "0x12345",
-				Author: "did:firefly:org/abcd",
-			},
-			Namespace: "ns1",
-			Node:      fftypes.NewUUID(),
-			Created:   fftypes.Now(),
-		},
-		Hash: fftypes.NewRandB32(),
-		TX: core.TransactionRef{
-			ID:   txid,
-			Type: core.TransactionTypeBatchPin,
-		},
-		Manifest: fftypes.JSONAnyPtr((&core.BatchManifest{
-			Messages: []*core.MessageManifestEntry{
-				{MessageRef: core.MessageRef{ID: msgID1}},
-				{MessageRef: core.MessageRef{ID: msgID2}},
-			},
-		}).String()),
-		Confirmed: fftypes.Now(),
-	}
-
-	// Rejects hash change
-	err = s.UpsertBatch(context.Background(), batchUpdated)
-	assert.Equal(t, database.HashMismatch, err)
-
-	batchUpdated.Hash = batch.Hash
-	err = s.UpsertBatch(context.Background(), batchUpdated)
+	// Try to insert again - should get back the existing row
+	existing, err = s.InsertOrGetBatch(ctx, batch)
 	assert.NoError(t, err)
-
-	// Check we get the exact same message back - note the removal of one of the batch elements
-	batchRead, err = s.GetBatchByID(ctx, "ns1", batchID)
-	assert.NoError(t, err)
-	batchJson, _ = json.Marshal(&batchUpdated)
-	batchReadJson, _ = json.Marshal(&batchRead)
-	assert.Equal(t, string(batchJson), string(batchReadJson))
+	assert.NotNil(t, existing)
 
 	// Query back the batch
 	fb := database.BatchQueryFactory.NewFilter(ctx)
 	filter := fb.And(
-		fb.Eq("id", batchUpdated.ID.String()),
-		fb.Eq("author", batchUpdated.Author),
+		fb.Eq("id", batch.ID.String()),
+		fb.Eq("author", batch.Author),
 		fb.Gt("created", "0"),
-		fb.Gt("confirmed", "0"),
 	)
 	batches, _, err := s.GetBatches(ctx, "ns1", filter)
 	assert.NoError(t, err)
@@ -137,7 +96,7 @@ func TestBatch2EWithDB(t *testing.T) {
 
 	// Negative test on filter
 	filter = fb.And(
-		fb.Eq("id", batchUpdated.ID.String()),
+		fb.Eq("id", batch.ID.String()),
 		fb.Eq("created", "0"),
 	)
 	batches, _, err = s.GetBatches(ctx, "ns1", filter)
@@ -152,7 +111,7 @@ func TestBatch2EWithDB(t *testing.T) {
 
 	// Test find updated value
 	filter = fb.And(
-		fb.Eq("id", batchUpdated.ID.String()),
+		fb.Eq("id", batch.ID.String()),
 		fb.Eq("author", author2),
 	)
 	batches, res, err := s.GetBatches(ctx, "ns1", filter.Count(true))
@@ -166,55 +125,30 @@ func TestBatch2EWithDB(t *testing.T) {
 func TestUpsertBatchFailBegin(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{})
+	_, err := s.InsertOrGetBatch(context.Background(), &core.BatchPersisted{})
 	assert.Regexp(t, "FF10114", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestUpsertBatchFailSelect(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	batchID := fftypes.NewUUID()
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
-	assert.Regexp(t, "FF10115", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestUpsertBatchFailInsert(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{}))
 	mock.ExpectExec("INSERT .*").WillReturnError(fmt.Errorf("pop"))
+	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{}))
 	mock.ExpectRollback()
 	batchID := fftypes.NewUUID()
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
+	_, err := s.InsertOrGetBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
 	assert.Regexp(t, "FF10116", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestUpsertBatchFailUpdate(t *testing.T) {
-	s, mock := newMockProvider().init()
-	batchID := fftypes.NewUUID()
-	hash := fftypes.NewRandB32()
-	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"hash"}).AddRow(hash))
-	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}, Hash: hash})
-	assert.Regexp(t, "FF10117", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestUpsertBatchFailCommit(t *testing.T) {
 	s, mock := newMockProvider().init()
 	batchID := fftypes.NewUUID()
 	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"id"}))
 	mock.ExpectExec("INSERT .*").WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit().WillReturnError(fmt.Errorf("pop"))
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
+	_, err := s.InsertOrGetBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
 	assert.Regexp(t, "FF10119", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }