Fix the spool's critical bugs before it ships

cmd/substreams/sink_sql_common.go

@@ -222,6 +222,7 @@ var fromProtoSchemaFlagNames = []string{
 	"disable-primary-keys",
 	"disable-unique-constraints",
 	"disable-block-number-index",
+	"disable-all-constraints",
 	"no-constraints",
 	"proto-file-override",
 }
@@ -277,14 +278,27 @@ func addFromProtoSchemaFlags(flags *pflag.FlagSet) {
 	flags.Bool("disable-block-number-index", false, "Leave out the index on _block_number_. Every table carries that column and every reorg deletes from every table by it, so without the index each undo is a sequential scan per table — a foreign key indexes its referenced side only. It is created when the sink starts, concurrently and outside the constraint pass: --apply-constraints describes the schema and is yours to schedule, where this one the sink depends on to undo a reorg. Measured over 10GiB it costs 2.6s to build and 2% of the table. Only dead weight on a run that can never reorg, such as --final-blocks-only.")
 	flags.String("proto-file-override", "", "Override protobuf file to use instead of extracting from substreams package")
 
-	flags.Bool("no-constraints", false, "Deprecated, use --disable-foreign-keys --disable-primary-keys=all --disable-unique-constraints=all.")
-	_ = flags.MarkDeprecated("no-constraints", "use --disable-foreign-keys --disable-primary-keys=all --disable-unique-constraints=all")
+	flags.Bool("disable-all-constraints", false, "Leave out every primary key, unique constraint and foreign key, the same as passing --disable-foreign-keys --disable-primary-keys=all --disable-unique-constraints=all together. The index on _block_number_ survives it: nothing in the annotations asks for that one, the reorg path does.")
+
+	flags.Bool("no-constraints", false, "Deprecated, use --disable-all-constraints.")
+	_ = flags.MarkDeprecated("no-constraints", "use --disable-all-constraints")
+}
+
+// addConstraintTimingFlag registers --apply-constraints, which says when the constraints
+// are created and so only means something to a command that loads rows.
+//
+// It is deliberately not on `setup`. There the question is not when but whether, since
+// `setup` either leaves the schema constrained or it does not, and two of the three values
+// would collapse onto the same answer. What `setup` creates is said by the --disable-*
+// flags instead.
+func addConstraintTimingFlag(flags *pflag.FlagSet) {
+	flags.String("apply-constraints", string(protosql.ConstraintsAuto), "When the schema's constraints are created: 'auto' has the sink create them once the backfill reaches chain HEAD or the end of a bounded run, 'manual' leaves it to the 'sink postgres constraints apply' command, 'always' creates them before the load. Creating them is a stop-the-world operation — indexes to build, foreign keys to validate, tables locked throughout — so on a large database 'manual' is how that pass goes into a maintenance window instead. Loading with them already in place is the expensive option: measured through binary COPY, 27x slower than loading without, where building the same constraints afterwards costs 3.3x. The index on _block_number_ is not one of these: the sink creates it when it starts, see --disable-block-number-index.")
 }
 
 // addFromProtoRunFlags registers the from-proto flags that only apply while the sink is
 // running. The ClickHouse specific ones are only registered for the clickhouse engine.
 func addFromProtoRunFlags(flags *pflag.FlagSet, driver string) {
-	flags.String("apply-constraints", string(protosql.ConstraintsAuto), "When the schema's constraints are created: 'auto' has the sink create them once the backfill reaches chain HEAD or the end of a bounded run, 'manual' leaves it to the 'sink postgres constraints apply' command, 'always' creates them before the load. Creating them is a stop-the-world operation — indexes to build, foreign keys to validate, tables locked throughout — so on a large database 'manual' is how that pass goes into a maintenance window instead. Loading with them already in place is the expensive option: measured through binary COPY, 27x slower than loading without, where building the same constraints afterwards costs 3.3x. The index on _block_number_ is not one of these: the sink creates it when it starts, see --disable-block-number-index.")
+	addConstraintTimingFlag(flags)
 
 	flags.String("write-mode", string(protosql.WriteModeAuto), "How a sealed spool segment reaches the database: 'copy' loads each table file with binary COPY, 'batch-insert' builds one multi-row INSERT per table, 'row-insert' issues one prepared INSERT per row. 'auto' picks copy on PostgreSQL, batch-insert on ClickHouse, and row-insert for a schema whose foreign keys cannot be ordered. An explicit mode the driver or the schema cannot support is an error rather than a downgrade. Ignored once the stream reaches chain HEAD, where the sink always inserts directly.")
 
@@ -882,6 +896,12 @@ func runFromProtoSetup(cmd *cobra.Command, driver, dsnString string, spkg *pbsub
 		return err
 	}
 
+	// `setup` creates the schema it is asked for, constraints included, and the --disable-*
+	// flags are what take them back out. There is no third answer here the way there is for
+	// a run, which can defer them until the backfill is over: this command creates the
+	// schema and exits, so a constraint it leaves out is one nothing will put back.
+	constraints.Timing = protosql.ConstraintsAlways
+
 	encoding, err := sinkBytesEncoding(cmd)
 	if err != nil {
 		return err
@@ -1184,9 +1204,9 @@ func fromProtoConstraintPolicy(cmd *cobra.Command) (protosql.ConstraintPolicy, e
 		PerTransaction:          intFlag(cmd, "constraints-per-transaction"),
 	}
 
-	// --no-constraints shipped in v1.21.0 and said "none of them at all", which is exactly
-	// what the three switches say together. It stays honoured for a release.
-	if flagChanged(cmd, "no-constraints") && boolFlag(cmd, "no-constraints") {
+	// --no-constraints shipped in v1.21.0 and said "none of them at all", which is what
+	// --disable-all-constraints says now. It stays honoured for a release.
+	if boolFlag(cmd, "disable-all-constraints") || (flagChanged(cmd, "no-constraints") && boolFlag(cmd, "no-constraints")) {
 		disabled := protosql.DisableAllConstraints()
 		policy.DisableForeignKeys = true
 		policy.DisablePrimaryKeys = disabled.DisablePrimaryKeys
@@ -1237,8 +1257,11 @@ func fromProtoSpoolOptions(cmd *cobra.Command) (*spool.Options, error) {
 		Dir:                 dir,
 		MaxBytes:            maxBytes,
 		WriteTargetDuration: sflags.MustGetDuration(cmd, "db-write-target-duration"),
-		SegmentMaxBytes:     segmentMaxBytes,
-		MaxIdle:             maxIdle,
+		// A segment is one indivisible database write, so keep its configured ceiling
+		// within the total spool budget rather than allowing an invalid combination to
+		// deadlock when the segment is sealed.
+		SegmentMaxBytes: min(segmentMaxBytes, maxBytes),
+		MaxIdle:         maxIdle,
 	}, nil
 }
 

cmd/substreams/sink_sql_common_test.go

@@ -3,7 +3,9 @@ package main
 import (
 	"testing"
 
+	"github.com/spf13/cobra"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // TestSinkUserAgent pins what the server sees: the two engines run the same commands, so
@@ -15,3 +17,49 @@ func TestSinkUserAgent(t *testing.T) {
 	assert.Equal(t, "sink_database_changes_pg", sinkUserAgent("sink_database_changes", sinkPostgresDriver))
 	assert.Equal(t, "sink_from_proto", sinkUserAgent("sink_from_proto", "duckdb"))
 }
+
+// TestFromProtoConstraintPolicyDisableAll covers the shorthand and the deprecated flag it
+// replaces, both of which have to mean exactly what the three switches mean together.
+func TestFromProtoConstraintPolicyDisableAll(t *testing.T) {
+	newCmd := func() *cobra.Command {
+		cmd := &cobra.Command{}
+		addFromProtoSchemaFlags(cmd.Flags())
+		addConstraintTimingFlag(cmd.Flags())
+
+		return cmd
+	}
+
+	t.Run("nothing disabled by default", func(t *testing.T) {
+		policy, err := fromProtoConstraintPolicy(newCmd())
+		require.NoError(t, err)
+		assert.False(t, policy.SkipsEverything())
+	})
+
+	t.Run("the shorthand disables all three", func(t *testing.T) {
+		cmd := newCmd()
+		require.NoError(t, cmd.Flags().Set("disable-all-constraints", "true"))
+
+		policy, err := fromProtoConstraintPolicy(cmd)
+		require.NoError(t, err)
+		assert.True(t, policy.SkipsEverything())
+	})
+
+	t.Run("the deprecated flag still means the same", func(t *testing.T) {
+		cmd := newCmd()
+		require.NoError(t, cmd.Flags().Set("no-constraints", "true"))
+
+		policy, err := fromProtoConstraintPolicy(cmd)
+		require.NoError(t, err)
+		assert.True(t, policy.SkipsEverything())
+	})
+}
+
+// TestSetupCommandsCarryNoConstraintTiming pins that --apply-constraints stays off `setup`:
+// there the question is whether the schema comes out constrained, not when, and two of the
+// flag's three values would collapse onto the same answer.
+func TestSetupCommandsCarryNoConstraintTiming(t *testing.T) {
+	for _, cmd := range []*cobra.Command{sinkPostgresSetupCmd, sinkClickhouseSetupCmd} {
+		assert.Nil(t, cmd.Flags().Lookup("apply-constraints"), "%s must not carry --apply-constraints", cmd.Use)
+		assert.NotNil(t, cmd.Flags().Lookup("disable-all-constraints"), "%s must carry --disable-all-constraints", cmd.Use)
+	}
+}

sink/sql/db_proto/sinker.go

@@ -61,6 +61,7 @@ func NewSinker(rootMessageDescriptor protoreflect.MessageDescriptor, sink *sink.
 func (s *Sinker) Run(ctx context.Context) error {
 	// Show stats one last time before exiting run
 	defer s.LogStats()
+	defer s.closeDatabase()
 
 	cursor, err := s.db.FetchCursor()
 	if err != nil {
@@ -104,6 +105,31 @@ func (s *Sinker) LogStats() {
 	s.stats.Log()
 }
 
+// closeTimeout bounds the shutdown seal. Writing the open segment's manifest and fsyncing
+// it is local work of a few milliseconds; anything past this is a disk that is not coming
+// back, and holding the process there helps nobody.
+const closeTimeout = 30 * time.Second
+
+// closeDatabase seals whatever is still held and releases the connections, on every way
+// out of the run rather than only on the one that ends a bounded range.
+//
+// A backfill is normally interrupted rather than completed — Ctrl-C, a killed pod, a
+// stream that errors — and not sealing the open segment on the way out throws away every
+// block in it, up to --db-write-max-size: recovery discards an unsealed segment, so those
+// blocks are streamed, and paid for, a second time. That is the outcome the spool exists
+// to prevent.
+//
+// It closes on a fresh context because the run's is already cancelled by the time an
+// interrupt reaches here, and the seal it has to perform is a write.
+func (s *Sinker) closeDatabase() {
+	ctx, cancel := context.WithTimeout(context.Background(), closeTimeout)
+	defer cancel()
+
+	if err := s.db.Close(ctx); err != nil {
+		s.logger.Warn("closing the database", zap.Error(err))
+	}
+}
+
 type Holder struct {
 	output *pbsubstreamsrpc.MapModuleOutput
 	data   *pbsubstreamsrpc.BlockScopedData
@@ -154,7 +180,7 @@ func (s *Sinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrp
 			}
 		}
 
-		if err := s.db.SwitchToDirectInserts(ctx, "stream reached the chain head"); err != nil {
+		if err := s.db.SwitchToDirectInserts(ctx, "stream reached the chain head", true); err != nil {
 			return fmt.Errorf("switching to direct inserts: %w", err)
 		}
 
@@ -208,7 +234,7 @@ func (s *Sinker) HandleBlockRangeCompletion(ctx context.Context, cursor *sink.Cu
 	// transactions while it is open. Draining it first is what keeps those blocks from
 	// being streamed, and paid for, twice, and what leaves the constraints to be created
 	// against a database that already holds every row of the range.
-	if err := s.db.SwitchToDirectInserts(ctx, "stream reached the end of the requested range"); err != nil {
+	if err := s.db.SwitchToDirectInserts(ctx, "stream reached the end of the requested range", false); err != nil {
 		return fmt.Errorf("draining before the end of the range: %w", err)
 	}
 

sink/sql/db_proto/sql/click_house/chapplier.go

@@ -80,7 +80,9 @@ func (a *chApplier) Apply(_ context.Context, dir string, manifest *spool.Manifes
 		return fmt.Errorf("parsing the cursor of a spooled segment: %w", err)
 	}
 
-	return a.database.StoreCursor(cursor)
+	// Straight to the file: StoreCursor would route it back into the spool, which is where
+	// this cursor came from.
+	return a.database.storeCursorFile(cursor)
 }
 
 func (a *chApplier) ordinal(table string) int {

sink/sql/db_proto/sql/click_house/chcodec_block_test.go

@@ -0,0 +1,23 @@
+package clickhouse
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestEncodeValuesRoundTripsABlockRow covers the row shape InsertBlock produces, now that
+// it goes through the spool like every other one. A tag missing here would fail the run at
+// spool time rather than silently, but only once a block is written — which is every block.
+func TestEncodeValuesRoundTripsABlockRow(t *testing.T) {
+	timestamp := time.Unix(1700000000, 0).UTC()
+	row := []any{uint64(42), "0xabc", timestamp, int64(7), false}
+
+	encoded, err := encodeValues(row)
+	require.NoError(t, err)
+
+	decoded, err := decodeValues(encoded)
+	require.NoError(t, err)
+	require.Equal(t, row, decoded)
+}

sink/sql/db_proto/sql/click_house/cursor_test.go

@@ -0,0 +1,35 @@
+package clickhouse
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/streamingfast/bstream"
+	sink "github.com/streamingfast/substreams/sink"
+	"github.com/stretchr/testify/require"
+)
+
+// TestStoreCursorFileWritesWhereFetchCursorReads pins the path the spool applier uses.
+//
+// StoreCursor records onto the open segment whenever a spool is active, so this is the
+// only thing that makes a spooled ClickHouse backfill resumable: routing the applier
+// through StoreCursor instead leaves the file untouched for the whole run.
+func TestStoreCursorFileWritesWhereFetchCursorReads(t *testing.T) {
+	database := &Database{cursorFilePath: filepath.Join(t.TempDir(), "cursor.txt")}
+
+	block := bstream.NewBlockRef("10a", 10)
+	opaque := (&bstream.Cursor{
+		Step:      bstream.StepNewIrreversible,
+		Block:     block,
+		HeadBlock: block,
+		LIB:       block,
+	}).ToOpaque()
+
+	cursor, err := sink.NewCursor(opaque)
+	require.NoError(t, err)
+	require.NoError(t, database.storeCursorFile(cursor))
+
+	read, err := database.FetchCursor()
+	require.NoError(t, err)
+	require.Equal(t, uint64(10), read.Block().Num())
+}

sink/sql/db_proto/sql/click_house/database.go

@@ -195,7 +195,7 @@ func (d *Database) clientNoCache(dsn *db.DSN) (*ch.Client, error) {
 // The spool trades freshness for throughput, which is what a backfill wants and the
 // opposite of what a sink at the chain head wants, where a block should be queryable when
 // it arrives rather than when the segment it lands in is full.
-func (d *Database) SwitchToDirectInserts(ctx context.Context, reason string) error {
+func (d *Database) SwitchToDirectInserts(ctx context.Context, reason string, _ bool) error {
 	if d.spool == nil {
 		return nil
 	}
@@ -330,11 +330,20 @@ func (d *Database) VerifySchemaCompatibility(ctx context.Context) error {
 // rowIDColumnByTable reports, for every table of the schema that exists in ClickHouse,
 // whether it carries the _row_id_ column. An absent schema yields an empty map rather
 // than an error: nothing is set up yet, so there is nothing to disagree with.
+//
+// It connects to 'default' rather than to the schema, for the same reason CreateDatabase
+// does: this runs before the database exists on a first run, and dialing one that is not
+// there never comes back — newClient retries a failed dial forever. system.columns is
+// global, so which database the connection names does not change the answer.
 func (d *Database) rowIDColumnByTable(ctx context.Context) (map[string]bool, error) {
-	client, err := d.client()
+	dsn := d.dsn.Clone()
+	dsn.Database = "default"
+
+	client, err := d.clientNoCache(dsn)
 	if err != nil {
 		return nil, fmt.Errorf("getting clickhouse client: %w", err)
 	}
+	defer client.Close()
 
 	var (
 		tables  chproto.ColStr
@@ -403,13 +412,10 @@ func (d *Database) Flush() (time.Duration, error) {
 
 	startFlush := time.Now()
 
-	// With a spool a flush only seals a segment once it is big enough; the write to
-	// ClickHouse happens later, on the applier's goroutine.
+	// With a spool the rows are already on disk and the write to ClickHouse happens later,
+	// on the applier's goroutine. The segment is sealed when the cursor covering it is
+	// recorded, which happens after this.
 	if d.spool != nil {
-		if err := d.spool.MaybeSeal(d.ctx); err != nil {
-			return 0, fmt.Errorf("sealing: %w", err)
-		}
-
 		return time.Since(startFlush), nil
 	}
 
@@ -424,9 +430,16 @@ func (d *Database) GetDialect() sql.Dialect {
 	return d.dialect
 }
 
+// InsertBlock writes the block row through the same path as every other row.
+//
+// It must not reach the accumulator directly: with a spool open that instance belongs to
+// the applier's goroutine, so appending to it from here is an unsynchronised write to the
+// map the applier swaps out on every flush. Going through Insert is also what tells the
+// spool which block the rows now being written belong to, without which a segment records
+// no block range at all.
 func (d *Database) InsertBlock(blockNum uint64, hash string, timestamp time.Time) error {
 	d.logger.Debug("inserting _block_", zap.Uint64("block_num", blockNum), zap.String("block_hash", hash))
-	err := d.inserter.insert("_blocks_", []any{blockNum, hash, timestamp, time.Now().UnixNano(), false})
+	err := d.Insert(sql.DialectTableBlock, []any{blockNum, hash, timestamp, time.Now().UnixNano(), false})
 	if err != nil {
 		return fmt.Errorf("inserting block %d: %w", blockNum, err)
 	}
@@ -512,9 +525,22 @@ func (d *Database) StoreCursor(cursor *sink.Cursor) error {
 	if d.spool != nil {
 		d.spool.RecordCursor(cursor.String())
 
-		return nil
+		// Sealed here rather than at flush, the cursor being the last thing a flush writes:
+		// sealing before it would close the segment on the cursor the previous flush stored,
+		// leaving it to claim a range its own cursor does not cover.
+		return d.spool.MaybeSeal(d.ctx)
 	}
 
+	return d.storeCursorFile(cursor)
+}
+
+// storeCursorFile writes the cursor where the next run reads it back.
+//
+// It is what the applier calls once a segment has reached the server. Going through
+// StoreCursor would hand the cursor straight back to the spool it just came out of —
+// leaving the file untouched for the whole backfill, and stamping an already-applied
+// cursor over the newer one on the segment still being written.
+func (d *Database) storeCursorFile(cursor *sink.Cursor) error {
 	if d.cursorFilePath == "" {
 		return fmt.Errorf("cursor file path is not set")
 	}

sink/sql/db_proto/sql/database.go

@@ -51,7 +51,11 @@ type Database interface {
 	// that never buffered. The reason is what the caller knows and the database does not
 	// — reaching the chain head, or reaching the end of a bounded range — and it is what
 	// the switch reports, so the log does not claim a head the stream never saw.
-	SwitchToDirectInserts(ctx context.Context, reason string) error
+	// atChainHead says the sink carries on from here rather than exiting, which is what
+	// makes the spool's own bookkeeping worth clearing: a run that stays live seals a
+	// segment or two on every restart and would otherwise accumulate their records for as
+	// long as it lives.
+	SwitchToDirectInserts(ctx context.Context, reason string, atChainHead bool) error
 	// WalkMessageDescriptorAndInsert reads the message through protoreflect only, so it
 	// does not care which dynamic implementation produced it — dynamicpb and hyperpb are
 	// both accepted, and the decoder picks.