refeactoring undo signal handling

streamingfast · Feb 1, 2024 · 6a02865 · 6a02865
1 parent 828d22b
commit 6a02865
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 12 deletions.
diff --git a/data/psql.go b/data/psql.go
@@ -89,14 +89,6 @@ func (p *Psql) handleTransaction(dbBlockID int64, transactionHash string) (dbTra
 	return
 }
 
-func (p *Psql) HandleBlockUndo(blockId string) error {
-	_, err := p.tx.Exec("DELETE CASCADE FROM solana_tokens.blocks WHERE hash = $1", blockId)
-	if err != nil {
-		return fmt.Errorf("deleting block: %w", err)
-	}
-	return nil
-}
-
 func (p *Psql) HandleInitializedAccount(dbBlockID int64, initializedAccounts []*pb.InitializedAccount) (err error) {
 	for _, initializedAccount := range initializedAccounts {
 		dbTransactionID, err := p.handleTransaction(dbBlockID, initializedAccount.TrxHash)
@@ -111,6 +103,14 @@ func (p *Psql) HandleInitializedAccount(dbBlockID int64, initializedAccounts []*
 	return nil
 }
 
+func (p *Psql) HandleBlocksUndo(lastValidBlockNum uint64) error {
+	_, err := p.tx.Exec("DELETE CASCADE FROM solana_tokens.blocks WHERE num > $1", lastValidBlockNum)
+	if err != nil {
+		return fmt.Errorf("deleting block from %s: %w", lastValidBlockNum, err)
+	}
+	return nil
+}
+
 var NotFound = errors.New("Not found")
 
 func (p *Psql) resolveAddress(derivedAddress string) (string, error) {

diff --git a/data/sinker.go b/data/sinker.go
@@ -149,15 +149,15 @@ func (s *Sinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrp
 }
 
 func (s *Sinker) HandleBlockUndoSignal(ctx context.Context, undoSignal *pbsubstreamsrpc.BlockUndoSignal, cursor *sink.Cursor) (err error) {
-	blockId := cursor.Block().ID()
+	lastValidBlockNum := undoSignal.LastValidBlock.Number
 
 	s.logger.Info("Handling undo block signal", zap.Stringer("block", cursor.Block()), zap.Stringer("cursor", cursor))
 
 	defer func() {
 		if err != nil {
 			if s.db.tx != nil {
 				e := s.db.RollbackTransaction()
-				err = fmt.Errorf("undo block: %s rollback transaction: %w: while handling err %w", blockId, e, err)
+				err = fmt.Errorf("undo blocks: %s rollback transaction: %w: while handling err %w", lastValidBlockNum, e, err)
 			}
 
 			return
@@ -172,9 +172,9 @@ func (s *Sinker) HandleBlockUndoSignal(ctx context.Context, undoSignal *pbsubstr
 		return fmt.Errorf("begin transaction: %w", err)
 	}
 
-	err = s.db.HandleBlockUndo(blockId)
+	err = s.db.HandleBlocksUndo(lastValidBlockNum)
 	if err != nil {
-		return fmt.Errorf("handle block %s undo: %w", blockId, err)
+		return fmt.Errorf("handle blocks undo from %d : %w", lastValidBlockNum, err)
 	}
 
 	err = s.db.StoreCursor(cursor)