fix critical reorg bug where only a single reversion would be perform…

…ed when handling a reorg

CHANGELOG.md

@@ -5,21 +5,32 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+
+## v4.0.0-rc.3
+
+* Fix a critical bug breaking the reorg management when more than one row needs to be reverted.
+
 ## v4.0.0-rc.2
 
+> :warning: This release candidate contains a critical bug in the reorg management and should not be used. Upgrade immediately to v4.0.0-rc.3
+
 * Support more networks with default mappings (ex: solana, soon: optimism, soon: bitcoin)
 * Add command "create-user <dsn> <username> <database>" to help creating more SQL users, read-only or otherwise
 * add 'enabled' field under "DBTConfig"
 * Removed PgwebFrontend and WireProtocolAccess fields from the SinkConfig message: they will now be deployed when on a development environment, so they are not mentionned here anymore.
 
 ## v4.0.0-rc.1
 
+> :warning: This release candidate contains a critical bug in the reorg management and should not be used. Upgrade immediately to v4.0.0-rc.3
+
 ### Fixes
 
 * Fix an issue preventing the `setup` command from running on a clickhouse backend because of the reorg settings.
 
 ## v4.0.0-beta
 
+> :warning: This release candidate contains a critical bug in the reorg management and should not be used. Upgrade immediately to v4.0.0-rc.3
+
 ### Highlights
 
 * This release brings support for managing reorgs in Postgres database, enabled by default when `--undo-buffer-size` to 0.

db/dialect_postgres.go

@@ -30,6 +30,7 @@ func (d postgresDialect) Revert(tx Tx, ctx context.Context, l *Loader, lastValid
 		return err
 	}
 
+	var reversions []func() error
 	l.logger.Info("reverting forked block block(s)", zap.Uint64("last_valid_final_block", lastValidFinalBlock))
 	if rows != nil { // rows will be nil with no error only in testing scenarios
 		defer rows.Close()
@@ -45,13 +46,23 @@ func (d postgresDialect) Revert(tx Tx, ctx context.Context, l *Loader, lastValid
 			l.logger.Debug("reverting", zap.String("operation", op), zap.String("table_name", table_name), zap.String("pk", pk), zap.Uint64("block_num", block_num))
 			prev_value := prev_value_nullable.String
 
-			if err := d.revertOp(tx, ctx, op, table_name, pk, prev_value, block_num); err != nil {
-				return fmt.Errorf("revertOp: %w", err)
-			}
+			// we can't call revertOp inside this loop, because it calls tx.ExecContext,
+			// which can't run while this query is "active" or it will silently discard the remaining rows!
+			reversions = append(reversions, func() error {
+				if err := d.revertOp(tx, ctx, op, table_name, pk, prev_value, block_num); err != nil {
+					return fmt.Errorf("revertOp: %w", err)
+				}
+				return nil
+			})
 		}
 		if err := rows.Err(); err != nil {
 			return fmt.Errorf("iterating on rows from query %q: %w", query, err)
 		}
+		for _, reversion := range reversions {
+			if err := reversion(); err != nil {
+				return fmt.Errorf("execution revert operation: %w", err)
+			}
+		}
 	}
 	pruneHistory := fmt.Sprintf(`DELETE FROM %s WHERE "block_num" > %d;`,
 		d.historyTable(l.schema),