Repository: hyperledger/firefly
Pull request #1047 (2022-09-07 14:19:03)

Properly enforce uniqueness of BatchPin blockchain events
Because listener_id is NULL for these events, we need separate indexes for when it is NULL and not NULL.


Patch:
diff --git a/auditevents/main.go b/auditevents/main.go
index 3f3b155a9..a9463bb09 100644
--- a/auditevents/main.go
+++ b/auditevents/main.go
@@ -85,17 +85,18 @@ func run(host string) error {
 		received = len(events)
 
 		for _, event := range events {
-			lastSequence = event.Sequence
 			listener := event.BlockchainEvent.Listener
 			src := event.BlockchainEvent.Source
 			if listener.String() != expectedListener || src != expectedSource {
 				continue
 			}
+			sequence := event.Sequence
 			protocolID := event.BlockchainEvent.ProtocolID
-			fmt.Printf("%-10d %s\n", lastSequence, protocolID)
+			fmt.Printf("%-10d %s\n", sequence, protocolID)
 			if protocolID <= lastProtocolID {
 				return fmt.Errorf("out of order events detected")
 			}
+			lastSequence = sequence
 			lastProtocolID = protocolID
 			validated++
 		}
diff --git a/db/migrations/postgres/000099_fix_blockchainevent_protocolid_index.down.sql b/db/migrations/postgres/000099_fix_blockchainevent_protocolid_index.down.sql
new file mode 100644
index 000000000..136e36400
--- /dev/null
+++ b/db/migrations/postgres/000099_fix_blockchainevent_protocolid_index.down.sql
@@ -0,0 +1,5 @@
+BEGIN;
+DROP INDEX blockchainevents_protocolid;
+DROP INDEX blockchainevents_listener_protocolid;
+CREATE UNIQUE INDEX blockchainevents_protocolid ON blockchainevents(namespace, listener_id, protocol_id);
+COMMIT;
diff --git a/db/migrations/postgres/000099_fix_blockchainevent_protocolid_index.up.sql b/db/migrations/postgres/000099_fix_blockchainevent_protocolid_index.up.sql
new file mode 100644
index 000000000..9fbc17118
--- /dev/null
+++ b/db/migrations/postgres/000099_fix_blockchainevent_protocolid_index.up.sql
@@ -0,0 +1,5 @@
+BEGIN;
+DROP INDEX blockchainevents_protocolid;
+CREATE UNIQUE INDEX blockchainevents_protocolid ON blockchainevents(namespace, protocol_id) WHERE listener_id IS NULL;
+CREATE UNIQUE INDEX blockchainevents_listener_protocolid ON blockchainevents(namespace, listener_id, protocol_id) WHERE listener_id IS NOT NULL;
+COMMIT;
diff --git a/db/migrations/sqlite/000099_fix_blockchainevent_protocolid_index.down.sql b/db/migrations/sqlite/000099_fix_blockchainevent_protocolid_index.down.sql
new file mode 100644
index 000000000..d3a32b045
--- /dev/null
+++ b/db/migrations/sqlite/000099_fix_blockchainevent_protocolid_index.down.sql
@@ -0,0 +1,3 @@
+DROP INDEX blockchainevents_protocolid;
+DROP INDEX blockchainevents_listener_protocolid;
+CREATE UNIQUE INDEX blockchainevents_protocolid ON blockchainevents(namespace, listener_id, protocol_id);
diff --git a/db/migrations/sqlite/000099_fix_blockchainevent_protocolid_index.up.sql b/db/migrations/sqlite/000099_fix_blockchainevent_protocolid_index.up.sql
new file mode 100644
index 000000000..4dc3f4ccf
--- /dev/null
+++ b/db/migrations/sqlite/000099_fix_blockchainevent_protocolid_index.up.sql
@@ -0,0 +1,3 @@
+DROP INDEX blockchainevents_protocolid;
+CREATE UNIQUE INDEX blockchainevents_protocolid ON blockchainevents(namespace, protocol_id) WHERE listener_id IS NULL;
+CREATE UNIQUE INDEX blockchainevents_listener_protocolid ON blockchainevents(namespace, listener_id, protocol_id) WHERE listener_id IS NOT NULL;


Test patch:
diff --git a/internal/database/sqlcommon/blockchainevents_sql_test.go b/internal/database/sqlcommon/blockchainevents_sql_test.go
index 3f73fc366..0be3e75f6 100644
--- a/internal/database/sqlcommon/blockchainevents_sql_test.go
+++ b/internal/database/sqlcommon/blockchainevents_sql_test.go
@@ -51,7 +51,7 @@ func TestBlockchainEventsE2EWithDB(t *testing.T) {
 		},
 	}
 
-	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBlockchainEvents, core.ChangeEventTypeCreated, "ns", event.ID).Return()
+	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBlockchainEvents, core.ChangeEventTypeCreated, "ns", event.ID).Return().Once()
 
 	_, err := s.InsertOrGetBlockchainEvent(ctx, event)
 	assert.NotNil(t, event.Timestamp)
@@ -83,7 +83,7 @@ func TestBlockchainEventsE2EWithDB(t *testing.T) {
 	eventReadJson, _ = json.Marshal(eventRead)
 	assert.Equal(t, string(eventJson), string(eventReadJson))
 
-	// Try to insert again with a new ID (should return existing row)
+	// Try to insert again with a new ID - should return existing row
 	event2 := &core.BlockchainEvent{
 		ID:         fftypes.NewUUID(),
 		Namespace:  event.Namespace,
@@ -94,6 +94,27 @@ func TestBlockchainEventsE2EWithDB(t *testing.T) {
 	existing, err := s.InsertOrGetBlockchainEvent(ctx, event2)
 	assert.NoError(t, err)
 	assert.Equal(t, event.ID, existing.ID)
+
+	// Try to insert an event twice (with nil listener) - should return existing row
+	event3 := &core.BlockchainEvent{
+		ID:         fftypes.NewUUID(),
+		Namespace:  "ns",
+		ProtocolID: "tx2",
+		Timestamp:  fftypes.Now(),
+	}
+	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBlockchainEvents, core.ChangeEventTypeCreated, "ns", event3.ID).Return().Once()
+	existing, err = s.InsertOrGetBlockchainEvent(ctx, event3)
+	assert.NoError(t, err)
+	assert.Nil(t, existing)
+	event4 := &core.BlockchainEvent{
+		ID:         fftypes.NewUUID(),
+		Namespace:  "ns",
+		ProtocolID: "tx2",
+		Timestamp:  fftypes.Now(),
+	}
+	existing, err = s.InsertOrGetBlockchainEvent(ctx, event4)
+	assert.NoError(t, err)
+	assert.Equal(t, event3.ID, existing.ID)
 }
 
 func TestInsertBlockchainEventFailBegin(t *testing.T) {