Merge pull request #488 from shutter-network/contract-event-index

Use event index from contract instead of local counter

Author: jannikluhn

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

@@ -11,11 +11,18 @@ INSERT INTO transaction_submitted_event (
     gas_limit
 )
 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-ON CONFLICT DO NOTHING;
+ON CONFLICT (index, eon) DO UPDATE SET
+block_number = $2,
+block_hash = $3,
+tx_index = $4,
+log_index = $5,
+identity_prefix = $7,
+sender = $8,
+gas_limit = $9;
 
 -- name: GetTransactionSubmittedEvents :many
 SELECT * FROM transaction_submitted_event
-WHERE eon = $1 AND index >= $2
+WHERE eon = $1 AND index >= $2 AND index < $2 + $3
 ORDER BY index ASC
 LIMIT $3;
 
@@ -27,16 +34,9 @@ SET block_hash = $1, block_number = $2, slot = $3;
 -- name: GetTransactionSubmittedEventsSyncedUntil :one
 SELECT * 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;
+SELECT max(index) + 1 FROM transaction_submitted_event
+WHERE eon = $1;
 
 -- name: GetTxPointer :one
 SELECT * FROM tx_pointer

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

@@ -219,7 +219,8 @@ func getTxPointer(ctx context.Context, db *pgxpool.Pool, eon int64, maxTxPointer
 			Int64("tx-pointer", txPointer).
 			Int64("tx-pointer-age", txPointerAge).
 			Msg("outdated tx pointer")
-		txPointer, err = gnosisKeyperDB.GetTransactionSubmittedEventCount(ctx, eon)
+		txPointerInt32, err := gnosisKeyperDB.GetTransactionSubmittedEventCount(ctx, eon)
+		txPointer = int64(txPointerInt32)
 		if err == pgx.ErrNoRows {
 			txPointer = 0
 		} else if err != nil {

rolling-shutter/keyperimpl/gnosis/newslot.go

@@ -48,9 +48,11 @@ func TestGetTxPointerInfiniteIntegration(t *testing.T) {
 		Value: 5,
 	})
 	assert.NilError(t, err)
-	err = db.SetTransactionSubmittedEventCount(ctx, gnosisDatabase.SetTransactionSubmittedEventCountParams{
-		Eon:        2,
-		EventCount: 10,
+	_, err = db.InsertTransactionSubmittedEvent(ctx, gnosisDatabase.InsertTransactionSubmittedEventParams{
+		Index:          9,
+		Eon:            2,
+		BlockHash:      []byte{},
+		IdentityPrefix: []byte{},
 	})
 	assert.NilError(t, err)
 
@@ -75,9 +77,11 @@ func TestGetTxPointerOutdatedIntegration(t *testing.T) {
 		Value: 5,
 	})
 	assert.NilError(t, err)
-	err = db.SetTransactionSubmittedEventCount(ctx, gnosisDatabase.SetTransactionSubmittedEventCountParams{
-		Eon:        2,
-		EventCount: 10,
+	_, err = db.InsertTransactionSubmittedEvent(ctx, gnosisDatabase.InsertTransactionSubmittedEventParams{
+		Index:          9,
+		Eon:            2,
+		BlockHash:      []byte{},
+		IdentityPrefix: []byte{},
 	})
 	assert.NilError(t, err)
 

rolling-shutter/keyperimpl/gnosis/newslot_test.go

@@ -152,22 +152,9 @@ func (s *SequencerSyncer) insertTransactionSubmittedEvents(
 	events []*sequencerBindings.SequencerTransactionSubmitted,
 ) error {
 	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,
+			Index:          int64(event.TxIndex),
 			BlockNumber:    int64(event.Raw.BlockNumber),
 			BlockHash:      event.Raw.BlockHash[:],
 			TxIndex:        int64(event.Raw.TxIndex),
@@ -180,25 +167,15 @@ func (s *SequencerSyncer) insertTransactionSubmittedEvents(
 		if err != nil {
 			return errors.Wrap(err, "failed to insert transaction submitted event into db")
 		}
-		metricsLatestTxSubmittedEventIndex.WithLabelValues(fmt.Sprint(event.Eon)).Set(float64(nextEventIndex))
-		nextEventIndices[event.Eon]++
+		metricsLatestTxSubmittedEventIndex.WithLabelValues(fmt.Sprint(event.Eon)).Set(float64(event.TxIndex))
 		log.Debug().
-			Int64("index", nextEventIndex).
+			Uint64("index", event.TxIndex).
 			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")
 	}
-	for eon, nextEventIndex := range nextEventIndices {
-		err := queries.SetTransactionSubmittedEventCount(ctx, database.SetTransactionSubmittedEventCountParams{
-			Eon:        int64(eon),
-			EventCount: nextEventIndex,
-		})
-		if err != nil {
-			return err
-		}
-	}
 	return nil
 }