Revert async flushes (#15)

* Revert "fix race condition"

This reverts commit 426df23.

* Revert "Improve metrics (#13)"

This reverts commit 5945758.

* Revert "Async flushes (#11)"

This reverts commit f8fd749.

Author: YaroShkvorets

cmd/substreams-sink-sql/common_flags.go

@@ -60,13 +60,12 @@ func newDBLoader(
 	batchBlockFlushInterval int,
 	batchRowFlushInterval int,
 	liveBlockFlushInterval int,
-	maxParallelFlushes int,
 	handleReorgs bool,
 ) (*db.Loader, error) {
 	moduleMismatchMode, err := db.ParseOnModuleHashMismatch(sflags.MustGetString(cmd, onModuleHashMistmatchFlag))
 	cli.NoError(err, "invalid mistmatch mode")
 
-	dbLoader, err := db.NewLoader(psqlDSN, batchBlockFlushInterval, batchRowFlushInterval, liveBlockFlushInterval, maxParallelFlushes, moduleMismatchMode, &handleReorgs, zlog, tracer)
+	dbLoader, err := db.NewLoader(psqlDSN, batchBlockFlushInterval, batchRowFlushInterval, liveBlockFlushInterval, moduleMismatchMode, &handleReorgs, zlog, tracer)
 	if err != nil {
 		return nil, fmt.Errorf("new psql loader: %w", err)
 	}
@@ -95,7 +94,7 @@ func AddCommonSinkerFlags(flags *pflag.FlagSet) {
 		- If 'warn' is used, it does the same as 'ignore' but it will log a warning message when it happens.
 		- If 'ignore' is set, we pick the cursor at the highest block number and use it as the starting point. Subsequent
 		updates to the cursor will overwrite the module hash in the database.
-	    `))
+	`))
 }
 
 func readBlockRangeArgument(in string) (blockRange *bstream.Range, err error) {

cmd/substreams-sink-sql/create_user.go

@@ -45,7 +45,7 @@ func createUserE(cmd *cobra.Command, args []string) error {
 	}
 
 	if err := retry(ctx, func(ctx context.Context) error {
-		dbLoader, err := db.NewLoader(dsn, 0, 0, 0, 1, db.OnModuleHashMismatchError, nil, zlog, tracer)
+		dbLoader, err := db.NewLoader(dsn, 0, 0, 0, db.OnModuleHashMismatchError, nil, zlog, tracer)
 		if err != nil {
 			return fmt.Errorf("new psql loader: %w", err)
 		}

cmd/substreams-sink-sql/generate_csv.go

@@ -107,15 +107,7 @@ func generateCsvE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("new base sinker: %w", err)
 	}
 
-	dbLoader, err := newDBLoader(
-		cmd,
-		dsn,
-		0,
-		0,
-		0,
-		1,
-		false,
-	)
+	dbLoader, err := newDBLoader(cmd, dsn, 0, 0, 0, false) // flush interval not used in CSV mode
 	if err != nil {
 		return fmt.Errorf("new db loader: %w", err)
 	}

cmd/substreams-sink-sql/run.go

@@ -32,7 +32,6 @@ var sinkRunCmd = Command(sinkRunE,
 		flags.Int("batch-row-flush-interval", 100_000, "When in catch up mode, flush every N rows or after batch-block-flush-interval, whichever comes first. Set to 0 to disable and only use batch-block-flush-interval. Ineffective if the sink is now in the live portion of the chain where only 'live-block-flush-interval' applies.")
 		flags.Int("live-block-flush-interval", 1, "When processing in live mode, flush every N blocks.")
 		flags.Int("flush-interval", 0, "(deprecated) please use --batch-block-flush-interval instead")
-		flags.Int("max-parallel-flushes", 1, "Maximum number of parallel async flushes. Minimum is 1.")
 		flags.String("idle-timeout", "", "Duration to wait without data messages before triggering a reconnect, e.g. '10m', '1h'. Waiting for first block doesn't count as idle.")
 		flags.StringP("endpoint", "e", "", "Specify the substreams endpoint, ex: `mainnet.eth.streamingfast.io:443`")
 	}),
@@ -106,15 +105,7 @@ func sinkRunE(cmd *cobra.Command, args []string) error {
 		batchBlockFlushInterval = sflags.MustGetInt(cmd, "flush-interval")
 	}
 
-	dbLoader, err := newDBLoader(
-		cmd,
-		dsn,
-		batchBlockFlushInterval,
-		sflags.MustGetInt(cmd, "batch-row-flush-interval"),
-		sflags.MustGetInt(cmd, "live-block-flush-interval"),
-		sflags.MustGetInt(cmd, "max-parallel-flushes"),
-		handleReorgs,
-	)
+	dbLoader, err := newDBLoader(cmd, dsn, batchBlockFlushInterval, sflags.MustGetInt(cmd, "batch-row-flush-interval"), sflags.MustGetInt(cmd, "live-block-flush-interval"), handleReorgs)
 	if err != nil {
 		return fmt.Errorf("new db loader: %w", err)
 	}

cmd/substreams-sink-sql/setup.go

@@ -46,7 +46,7 @@ func sinkSetupE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("extract sink config: %w", err)
 	}
 
-	dbLoader, err := db.NewLoader(dsn, 0, 0, 0, 1, db.OnModuleHashMismatchError, nil, zlog, tracer)
+	dbLoader, err := db.NewLoader(dsn, 0, 0, 0, db.OnModuleHashMismatchError, nil, zlog, tracer)
 	if err != nil {
 		return fmt.Errorf("new psql loader: %w", err)
 	}

cmd/substreams-sink-sql/tools.go

@@ -145,7 +145,7 @@ func toolsDeleteCursorE(cmd *cobra.Command, args []string) error {
 
 func toolsCreateLoader() *db.Loader {
 	dsn := viper.GetString("tools-global-dsn")
-	loader, err := db.NewLoader(dsn, 0, 0, 0, 1, db.OnModuleHashMismatchIgnore, nil, zlog, tracer)
+	loader, err := db.NewLoader(dsn, 0, 0, 0, db.OnModuleHashMismatchIgnore, nil, zlog, tracer)
 	cli.NoError(err, "Unable to instantiate database manager from DSN %q", dsn)
 
 	if err := loader.LoadTables(); err != nil {

db/db.go

@@ -4,14 +4,10 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"strings"
 	"time"
 
-	"sync"
-
 	"github.com/jimsmart/schema"
 	"github.com/streamingfast/logging"
-	sink "github.com/streamingfast/substreams-sink"
 	orderedmap "github.com/wk8/go-ordered-map/v2"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -26,43 +22,6 @@ type OrderedMap[K comparable, V any] struct {
 	*orderedmap.OrderedMap[K, V]
 }
 
-// newFlushLoader creates a minimal Loader used exclusively to perform a flush
-// using the provided entries snapshot. It intentionally does not copy
-// synchronization primitives.
-func (l *Loader) newFlushLoader(snapshot *OrderedMap[string, *OrderedMap[string, *Operation]]) *Loader {
-	return &Loader{
-		DB:                       l.DB,
-		database:                 l.database,
-		schema:                   l.schema,
-		entries:                  snapshot,
-		entriesCount:             l.entriesCount,
-		tables:                   l.tables,
-		cursorTable:              l.cursorTable,
-		handleReorgs:             l.handleReorgs,
-		batchBlockFlushInterval:  l.batchBlockFlushInterval,
-		batchRowFlushInterval:    l.batchRowFlushInterval,
-		liveBlockFlushInterval:   l.liveBlockFlushInterval,
-		moduleMismatchMode:       l.moduleMismatchMode,
-		maxFlushRetries:          l.maxFlushRetries,
-		sleepBetweenFlushRetries: l.sleepBetweenFlushRetries,
-		logger:                   l.logger,
-		tracer:                   l.tracer,
-		testTx:                   l.testTx,
-	}
-}
-
-// SetOnFlush sets an optional observer invoked on successful flush completion.
-// The callback receives the number of rows flushed and the flush duration.
-func (l *Loader) SetOnFlush(cb func(rows int, dur time.Duration)) {
-	l.onFlush = cb
-}
-
-// SetOnFlushError sets an optional observer invoked when an async flush fails.
-// This enables the caller to handle shutdown or retries.
-func (l *Loader) SetOnFlushError(cb func(err error)) {
-	l.onFlushError = cb
-}
-
 func NewOrderedMap[K comparable, V any]() *OrderedMap[K, V] {
 	return &OrderedMap[K, V]{OrderedMap: orderedmap.New[K, V]()}
 }
@@ -93,27 +52,13 @@ type Loader struct {
 	tracer logging.Tracer
 
 	testTx *TestTx // used for testing: if non-nil, 'loader.BeginTx()' will return this object instead of a real *sql.Tx
-
-	// async flush
-	cond               *sync.Cond
-	activeFlushes      int
-	maxParallelFlushes int
-
-	// onFlush is an optional observer called when a flush completes successfully.
-	// It receives the number of rows flushed and the total duration of the flush.
-	onFlush func(rows int, dur time.Duration)
-
-	// onFlushError is an optional observer invoked when an async flush fails.
-	// If set, it should trigger appropriate shutdown from the caller side.
-	onFlushError func(err error)
 }
 
 func NewLoader(
 	psqlDsn string,
 	batchBlockFlushInterval int,
 	batchRowFlushInterval int,
 	liveBlockFlushInterval int,
-	maxParallelFlushes int,
 	moduleMismatchMode OnModuleHashMismatch,
 	handleReorgs *bool,
 	logger *zap.Logger,
@@ -129,10 +74,6 @@ func NewLoader(
 		return nil, fmt.Errorf("open db connection: %w", err)
 	}
 
-	if maxParallelFlushes < 1 {
-		maxParallelFlushes = 1
-	}
-
 	l := &Loader{
 		DB:                       db,
 		database:                 dsn.database,
@@ -147,9 +88,7 @@ func NewLoader(
 		moduleMismatchMode:       moduleMismatchMode,
 		logger:                   logger,
 		tracer:                   tracer,
-		maxParallelFlushes:       maxParallelFlushes,
 	}
-	l.cond = sync.NewCond(&sync.Mutex{})
 	_, err = l.tryDialect()
 	if err != nil {
 		return nil, fmt.Errorf("dialect not found: %s", err)
@@ -170,7 +109,6 @@ func NewLoader(
 		zap.Int("batch_block_flush_interval", batchBlockFlushInterval),
 		zap.Int("batch_row_flush_interval", batchRowFlushInterval),
 		zap.Int("live_block_flush_interval", liveBlockFlushInterval),
-		zap.Int("max_parallel_flushes", maxParallelFlushes),
 		zap.String("driver", dsn.driver),
 		zap.String("database", dsn.database),
 		zap.String("schema", dsn.schema),
@@ -212,28 +150,7 @@ func (l *Loader) LiveBlockFlushInterval() int {
 	return l.liveBlockFlushInterval
 }
 
-func (l *Loader) GetPrimaryKey(tableName string, pk string) (map[string]string, error) {
-	table, found := l.tables[tableName]
-	if !found {
-		return nil, fmt.Errorf("unknown table %q", tableName)
-	}
-	if len(table.primaryColumns) == 1 {
-		return map[string]string{table.primaryColumns[0].name: pk}, nil
-	}
-	parts := strings.Split(pk, "/")
-	if len(parts) != len(table.primaryColumns) {
-		return nil, fmt.Errorf("composite primary key value count mismatch for table %q: got %d parts, expected %d", tableName, len(parts), len(table.primaryColumns))
-	}
-	res := make(map[string]string, len(parts))
-	for i, col := range table.primaryColumns {
-		res[col.name] = parts[i]
-	}
-	return res, nil
-}
-
 func (l *Loader) FlushNeeded() bool {
-	l.cond.L.Lock()
-	defer l.cond.L.Unlock()
 	totalRows := 0
 	// todo keep a running count when inserting/deleting rows directly
 	for pair := l.entries.Oldest(); pair != nil; pair = pair.Next() {
@@ -242,72 +159,6 @@ func (l *Loader) FlushNeeded() bool {
 	return totalRows > l.batchRowFlushInterval
 }
 
-// WaitForAllFlushes blocks until there are no in-flight async flushes.
-func (l *Loader) WaitForAllFlushes() {
-	l.cond.L.Lock()
-	defer l.cond.L.Unlock()
-
-	for l.activeFlushes > 0 {
-		l.cond.Wait()
-	}
-}
-
-// FlushAsync triggers a non-blocking flush. Blocks if the maximum number of parallel flushes is reached until a flush is completed.
-func (l *Loader) FlushAsync(ctx context.Context, outputModuleHash string, cursor *sink.Cursor, lastFinalBlock uint64) {
-	l.logger.Debug("async flush: starting flush", zap.Int("active_flushes", l.activeFlushes), zap.Uint64("last_final_block", lastFinalBlock))
-	l.cond.L.Lock()
-	for l.activeFlushes >= l.maxParallelFlushes {
-		l.logger.Debug("async flush: maximum number of parallel flushes reached, waiting for a flush to complete")
-		l.cond.Wait()
-	}
-	// Snapshot entries and replace with a fresh buffer
-	snapshot := l.entries
-	l.entries = NewOrderedMap[string, *OrderedMap[string, *Operation]]()
-	l.activeFlushes++
-
-	// Build a lightweight loader for flushing while still under lock to avoid racy reads of fields.
-	flushLoader := l.newFlushLoader(snapshot)
-
-	l.cond.L.Unlock()
-
-	l.logger.Debug("async flush started", zap.Int("active_flushes", l.activeFlushes))
-
-	go func() {
-		// cleanup defer
-		defer func() {
-			l.cond.L.Lock()
-			l.activeFlushes--
-			l.cond.Broadcast()
-			l.cond.L.Unlock()
-		}()
-
-		// Disallow cancellation of the context to prevent holes in the data with parallel flushes
-		if l.maxParallelFlushes > 1 {
-			ctx = context.WithoutCancel(ctx)
-		}
-
-		start := time.Now()
-		rowFlushedCount, err := flushLoader.Flush(ctx, outputModuleHash, cursor, lastFinalBlock)
-		if err != nil {
-			l.logger.Warn("async flush failed after retries", zap.Error(err))
-			if l.onFlushError != nil {
-				l.onFlushError(err)
-			}
-			return
-		}
-
-		took := time.Since(start)
-		l.logger.Debug("async flush complete",
-			zap.Int("row_count", rowFlushedCount),
-			zap.Duration("took", took))
-
-		// Notify observer outside the lock
-		if l.onFlush != nil {
-			l.onFlush(rowFlushedCount, took)
-		}
-	}()
-}
-
 func (l *Loader) LoadTables() error {
 	schemaTables, err := schema.Tables(l.DB)
 	if err != nil {
@@ -453,13 +304,6 @@ func (l *Loader) HasTable(tableName string) bool {
 	return false
 }
 
-// GetActiveFlushesNum returns the current number of in-flight async flushes.
-func (l *Loader) GetActiveFlushesNum() int {
-	l.cond.L.Lock()
-	defer l.cond.L.Unlock()
-	return l.activeFlushes
-}
-
 func (l *Loader) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
 	encoder.AddUint64("entries_count", l.entriesCount)
 	return nil

db/flush.go

@@ -22,9 +22,6 @@ func (l *Loader) Flush(ctx context.Context, outputModuleHash string, cursor *sin
 				zap.Int("row_count", rowFlushedCount),
 				zap.Duration("took", time.Since(startAt)),
 			)
-			if l.onFlush != nil {
-				l.onFlush(rowFlushedCount, time.Since(startAt))
-			}
 			return rowFlushedCount, nil
 		}
 		l.logger.Warn("Flush attempt failed",

db/operations_test.go

@@ -19,7 +19,7 @@ func TestEscapeColumns(t *testing.T) {
 		t.Skip(`PG_DSN not set, please specify PG_DSN to run this test, example: PG_DSN="psql://dev-node:insecure-change-me-in-prod@localhost:5432/dev-node?enable_incremental_sort=off&sslmode=disable"`)
 	}
 
-	dbLoader, err := NewLoader(dsn, 0, 0, 0, 1, OnModuleHashMismatchIgnore, nil, zlog, tracer)
+	dbLoader, err := NewLoader(dsn, 0, 0, 0, OnModuleHashMismatchIgnore, nil, zlog, tracer)
 	require.NoError(t, err)
 
 	tx, err := dbLoader.DB.Begin()
@@ -68,7 +68,7 @@ func TestEscapeValues(t *testing.T) {
 		t.Skip(`PG_DSN not set, please specify PG_DSN to run this test, example: PG_DSN="psql://dev-node:insecure-change-me-in-prod@localhost:5432/dev-node?enable_incremental_sort=off&sslmode=disable"`)
 	}
 
-	dbLoader, err := NewLoader(dsn, 0, 0, 0, 1, OnModuleHashMismatchIgnore, nil, zlog, tracer)
+	dbLoader, err := NewLoader(dsn, 0, 0, 0, OnModuleHashMismatchIgnore, nil, zlog, tracer)
 	require.NoError(t, err)
 
 	tx, err := dbLoader.DB.Begin()