Store index of transaction submitted events

Author: jannikluhn

rolling-shutter/keyperimpl/gnosis/database/gnosiskeyper.sqlc.gen.go

@@ -1,5 +1,6 @@
 -- name: InsertTransactionSubmittedEvent :execresult
 INSERT INTO transaction_submitted_event (
+    index,
     block_number,
     block_hash,
     tx_index,
@@ -9,7 +10,7 @@ INSERT INTO transaction_submitted_event (
     sender,
     gas_limit
 )
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
 ON CONFLICT DO NOTHING;
 
 -- name: SetTransactionSubmittedEventsSyncedUntil :exec
@@ -20,6 +21,17 @@ SET block_number = $1;
 -- name: GetTransactionSubmittedEventsSyncedUntil :one
 SELECT block_number FROM transaction_submitted_events_synced_until LIMIT 1;
 
+-- name: SetTransactionSubmittedEventCount :exec
+INSERT INTO transaction_submitted_event_count (eon, event_count)
+VALUES ($1, $2)
+ON CONFLICT (eon) DO UPDATE
+SET event_count = $2;
+
+-- name: GetTransactionSubmittedEventCount :one
+SELECT event_count FROM transaction_submitted_event_count
+WHERE eon = $1
+LIMIT 1;
+
 -- name: GetLocalTxPointer :one
 SELECT local, local_block FROM tx_pointer LIMIT 1;
 

rolling-shutter/keyperimpl/gnosis/database/models.sqlc.gen.go

@@ -3,20 +3,25 @@
 -- the schema. We'll use this to check we're using the right schema.
 
 CREATE TABLE transaction_submitted_event (
-    block_number bigint CHECK (block_number >= 0),
-    block_hash bytea,
-    tx_index bigint CHECK (tx_index >= 0),
-    log_index bigint CHECK (log_index >= 0),
+    index bigint PRIMARY KEY CHECK (index >= 0),
+    block_number bigint NOT NULL CHECK (block_number >= 0),
+    block_hash bytea NOT NULL,
+    tx_index bigint NOT NULL CHECK (tx_index >= 0),
+    log_index bigint NOT NULL CHECK (log_index >= 0),
     eon bigint NOT NULL CHECK (eon >= 0),
     identity_prefix bytea NOT NULL,
     sender text NOT NULL,
-    gas_limit bigint NOT NULL CHECK (gas_limit >= 0),
-    PRIMARY KEY (block_number, block_hash, tx_index, log_index)
+    gas_limit bigint NOT NULL CHECK (gas_limit >= 0)
 );
 
 CREATE TABLE transaction_submitted_events_synced_until(
     enforce_one_row bool PRIMARY KEY DEFAULT true,
-    block_number bigint NOT NULL CHECK (block_number > 0)
+    block_number bigint NOT NULL CHECK (block_number >= 0)
+);
+
+CREATE TABLE transaction_submitted_event_count(
+    eon bigint PRIMARY KEY,
+    event_count bigint NOT NULL DEFAULT 0 CHECK (event_count >= 0)
 );
 
 -- tx_pointer stores what we know about the current value of the tx pointer. There are two

rolling-shutter/keyperimpl/gnosis/database/sql/queries/gnosiskeyper.sql

@@ -33,7 +33,7 @@ func (s *SequencerSyncer) Sync(ctx context.Context, block uint64) error {
 	if err == pgx.ErrNoRows {
 		start = 0
 	} else {
-		start = uint64(syncedUntilBlock) + 1
+		start = uint64(syncedUntilBlock + 1)
 	}
 
 	log.Debug().
@@ -69,19 +69,56 @@ func (s *SequencerSyncer) Sync(ctx context.Context, block uint64) error {
 	if it.Error() != nil {
 		return errors.Wrap(it.Error(), "failed to iterate transaction submitted events")
 	}
-
 	if len(events) == 0 {
 		log.Debug().
 			Uint64("start-block", start).
 			Uint64("end-block", block).
 			Msg("no transaction submitted events found")
-		return nil
 	}
 
 	return s.DBPool.BeginFunc(ctx, func(tx pgx.Tx) error {
+		err = s.insertTransactionSubmittedEvents(ctx, tx, events)
+		if err != nil {
+			return err
+		}
+
+		newSyncedUntilBlock, err := medley.Uint64ToInt64Safe(block)
+		if err != nil {
+			return err
+		}
+		err = queries.SetTransactionSubmittedEventsSyncedUntil(ctx, newSyncedUntilBlock)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
+// insertTransactionSubmittedEvents inserts the given events into the database and updates the
+// transaction submitted event number accordingly.
+func (s *SequencerSyncer) insertTransactionSubmittedEvents(
+	ctx context.Context,
+	tx pgx.Tx,
+	events []*sequencerBindings.SequencerTransactionSubmitted,
+) error {
+	if len(events) > 0 {
 		queries := database.New(tx)
+		nextEventIndices := make(map[uint64]int64)
 		for _, event := range events {
+			nextEventIndex, ok := nextEventIndices[event.Eon]
+			if !ok {
+				nextEventIndexFromDB, err := queries.GetTransactionSubmittedEventCount(ctx, int64(event.Eon))
+				if err == pgx.ErrNoRows {
+					nextEventIndexFromDB = 0
+				} else if err != nil {
+					return errors.Wrapf(err, "failed to query count of transaction submitted events for eon %d", event.Eon)
+				}
+				nextEventIndices[event.Eon] = nextEventIndexFromDB
+				nextEventIndex = nextEventIndexFromDB
+			}
+
 			_, err := queries.InsertTransactionSubmittedEvent(ctx, database.InsertTransactionSubmittedEventParams{
+				Index:          nextEventIndex,
 				BlockNumber:    int64(event.Raw.BlockNumber),
 				BlockHash:      event.Raw.BlockHash[:],
 				TxIndex:        int64(event.Raw.TxIndex),
@@ -94,18 +131,25 @@ func (s *SequencerSyncer) Sync(ctx context.Context, block uint64) error {
 			if err != nil {
 				return errors.Wrap(err, "failed to insert transaction submitted event into db")
 			}
+			nextEventIndices[event.Eon]++
 			log.Debug().
+				Int64("index", nextEventIndex).
 				Uint64("block", event.Raw.BlockNumber).
 				Uint64("eon", event.Eon).
 				Hex("identityPrefix", event.IdentityPrefix[:]).
 				Hex("sender", event.Sender.Bytes()).
 				Uint64("gasLimit", event.GasLimit.Uint64()).
 				Msg("synced new transaction submitted event")
 		}
-		newSyncedUntilBlock, err := medley.Uint64ToInt64Safe(block)
-		if err != nil {
-			return err
+		for eon, nextEventIndex := range nextEventIndices {
+			err := queries.SetTransactionSubmittedEventCount(ctx, database.SetTransactionSubmittedEventCountParams{
+				Eon:        int64(eon),
+				EventCount: nextEventIndex,
+			})
+			if err != nil {
+				return err
+			}
 		}
-		return queries.SetTransactionSubmittedEventsSyncedUntil(ctx, newSyncedUntilBlock)
-	})
+	}
+	return nil
 }