diff --git a/cmd/substreams/sink_sql_common.go b/cmd/substreams/sink_sql_common.go
index c6f476740..0906b9e54 100644
--- a/cmd/substreams/sink_sql_common.go
+++ b/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
 }
 
diff --git a/cmd/substreams/sink_sql_common_test.go b/cmd/substreams/sink_sql_common_test.go
index e1029fb8b..8d90cdd83 100644
--- a/cmd/substreams/sink_sql_common_test.go
+++ b/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)
+	}
+}
diff --git a/sink/sql/db_proto/sinker.go b/sink/sql/db_proto/sinker.go
index e6350d58a..60bf0023a 100644
--- a/sink/sql/db_proto/sinker.go
+++ b/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)
 	}
 
diff --git a/sink/sql/db_proto/sql/click_house/chapplier.go b/sink/sql/db_proto/sql/click_house/chapplier.go
index cb595f818..35a97b3b4 100644
--- a/sink/sql/db_proto/sql/click_house/chapplier.go
+++ b/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 {
diff --git a/sink/sql/db_proto/sql/click_house/chcodec_block_test.go b/sink/sql/db_proto/sql/click_house/chcodec_block_test.go
new file mode 100644
index 000000000..a70262e2a
--- /dev/null
+++ b/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)
+}
diff --git a/sink/sql/db_proto/sql/click_house/cursor_test.go b/sink/sql/db_proto/sql/click_house/cursor_test.go
new file mode 100644
index 000000000..6bfce745f
--- /dev/null
+++ b/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())
+}
diff --git a/sink/sql/db_proto/sql/click_house/database.go b/sink/sql/db_proto/sql/click_house/database.go
index b0d8b4d24..1290691fa 100644
--- a/sink/sql/db_proto/sql/click_house/database.go
+++ b/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")
 	}
diff --git a/sink/sql/db_proto/sql/database.go b/sink/sql/db_proto/sql/database.go
index f32d0dce8..4426da60c 100644
--- a/sink/sql/db_proto/sql/database.go
+++ b/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.
diff --git a/sink/sql/db_proto/sql/dialect.go b/sink/sql/db_proto/sql/dialect.go
index 39030e4d2..597ca99d5 100644
--- a/sink/sql/db_proto/sql/dialect.go
+++ b/sink/sql/db_proto/sql/dialect.go
@@ -105,13 +105,7 @@ func (d *BaseDialect) AddForeignKeyReferencing(table string, referencedTable str
 // order of rows within one table, which no table-level ordering can address — so those
 // edges are skipped.
 func (d *BaseDialect) TableApplyOrder() ([]string, error) {
-	// The block table is referenced by every other one and referenced by nothing, so it
-	// always leads.
-	names := []string{DialectTableBlock}
-	for name := range d.TableRegistry {
-		names = append(names, name)
-	}
-	sort.Strings(names[1:])
+	names := d.TableNames()
 
 	known := make(map[string]bool, len(names))
 	for _, name := range names {
@@ -167,6 +161,21 @@ func (d *BaseDialect) TableApplyOrder() ([]string, error) {
 	return ordered, nil
 }
 
+// TableNames lists every table of the schema in a stable order: the block table first,
+// being referenced by every other one and referencing none, then the rest sorted.
+//
+// It is what TableApplyOrder starts from, and what the reorg path falls back to when no
+// foreign key order exists.
+func (d *BaseDialect) TableNames() []string {
+	names := []string{DialectTableBlock}
+	for name := range d.TableRegistry {
+		names = append(names, name)
+	}
+	sort.Strings(names[1:])
+
+	return names
+}
+
 // TableApplyRanks is TableApplyOrder as a lookup, for sorting a set of tables that is not
 // the whole schema. A table the dialect does not know about ranks last.
 func (d *BaseDialect) TableApplyRanks() (map[string]int, error) {
diff --git a/sink/sql/db_proto/sql/dialect_order_test.go b/sink/sql/db_proto/sql/dialect_order_test.go
index 138f5285c..104e1eadc 100644
--- a/sink/sql/db_proto/sql/dialect_order_test.go
+++ b/sink/sql/db_proto/sql/dialect_order_test.go
@@ -157,3 +157,18 @@ func TestTableApplyOrder(t *testing.T) {
 		}
 	})
 }
+
+// TestTableNamesCoversEverySchemaTable is what the reorg path falls back to when the
+// foreign keys cannot be ordered. It has to name every table, or an undo would leave rows
+// of the ones it skipped behind.
+func TestTableNamesCoversEverySchemaTable(t *testing.T) {
+	dialect := dialectWith(
+		[]string{"orders", "customers"},
+		map[string]string{"orders": "customers", "customers": "orders"},
+	)
+
+	_, err := dialect.TableApplyOrder()
+	require.Error(t, err, "the schema is cyclic on purpose")
+
+	assert.Equal(t, []string{DialectTableBlock, "customers", "orders"}, dialect.TableNames())
+}
diff --git a/sink/sql/db_proto/sql/postgres/database.go b/sink/sql/db_proto/sql/postgres/database.go
index 7791c5c5c..a7920d3fb 100644
--- a/sink/sql/db_proto/sql/postgres/database.go
+++ b/sink/sql/db_proto/sql/postgres/database.go
@@ -282,7 +282,7 @@ func (d *Database) openDirectInserter() error {
 //
 // Everything spooled is applied before the switch, so no row is left behind, and the
 // spool is closed for good — a stream that has reached the head does not go back.
-func (d *Database) SwitchToDirectInserts(ctx context.Context, reason string) error {
+func (d *Database) SwitchToDirectInserts(ctx context.Context, reason string, atChainHead bool) error {
 	inserter, ok := d.inserter.(*localBufferInserter)
 	if !ok {
 		return nil
@@ -296,6 +296,18 @@ func (d *Database) SwitchToDirectInserts(ctx context.Context, reason string) err
 		return fmt.Errorf("draining the spool: %w", err)
 	}
 
+	if atChainHead {
+		// The drain above applied every segment and removed its directory, so there is
+		// nothing left on disk for a record to answer for. Only a run that carries on past
+		// here bothers: it seals a segment or two on each restart before reaching the head
+		// again, and those records would otherwise pile up for the life of the sink.
+		if err := inserter.applier.clearSegments(ctx); err != nil {
+			// Bookkeeping, and the spool is already drained: the rows are safe either way,
+			// so this is not worth failing the switch over.
+			d.logger.Warn("could not clear the applied-segment records", zap.Error(err))
+		}
+	}
+
 	// The spool is what copy mode is; without it the direct path is the multi-row one.
 	if d.resolvedWriteMode == sql.WriteModeCopy {
 		d.resolvedWriteMode = sql.WriteModeBatchInsert
@@ -968,7 +980,15 @@ func (d *Database) HandleBlocksUndo(lastValidBlockNum uint64) (err error) {
 	// parent-child keys and quietly get sibling references wrong.
 	ordered, err := d.dialect.TableApplyOrder()
 	if err != nil {
-		return err
+		// A cyclic foreign key graph has no such order, and refusing here would leave the
+		// sink unable to undo a reorg — or even to start, since Run undoes from the stored
+		// cursor — on exactly the schema row-insert mode exists to support. The deletes all
+		// run in one transaction and key on _block_number_ alone, so a stable arbitrary
+		// order is right wherever the references are absent, and where a cycle really is
+		// enforced no order would have worked either.
+		d.logger.Debug("the schema's tables cannot be ordered by their foreign keys, undoing in schema order",
+			zap.Error(err))
+		ordered = d.dialect.TableNames()
 	}
 
 	var rowsAffected int64
diff --git a/sink/sql/db_proto/sql/postgres/local_buffer_inserter.go b/sink/sql/db_proto/sql/postgres/local_buffer_inserter.go
index 7cecbe5e1..1822ea0d2 100644
--- a/sink/sql/db_proto/sql/postgres/local_buffer_inserter.go
+++ b/sink/sql/db_proto/sql/postgres/local_buffer_inserter.go
@@ -19,7 +19,10 @@ import (
 // the stream stops waiting on PostgreSQL.
 type localBufferInserter struct {
 	buffer *spool.Spool
-	logger *zap.Logger
+	// applier is kept so the switch to direct inserts can clear the bookkeeping it owns
+	// once the spool it belongs to has been drained.
+	applier *pgApplier
+	logger  *zap.Logger
 }
 
 func newLocalBufferInserter(ctx context.Context, database *Database, format spool.Format, options spool.Options, logger *zap.Logger) (*localBufferInserter, error) {
@@ -51,7 +54,7 @@ func newLocalBufferInserter(ctx context.Context, database *Database, format spoo
 		return nil, err
 	}
 
-	return &localBufferInserter{buffer: buf, logger: logger.Named("spool_inserter")}, nil
+	return &localBufferInserter{buffer: buf, applier: applier, logger: logger.Named("spool_inserter")}, nil
 }
 
 // loadColumnLayouts resolves every table's column order and type OIDs from the live
@@ -88,7 +91,12 @@ func (i *localBufferInserter) insert(table string, values []any, database *Datab
 			return fmt.Errorf("expected a string cursor, got %T", values[1])
 		}
 		i.buffer.RecordCursor(cursor)
-		return nil
+
+		// Sealing happens here rather than at flush, because the cursor is 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. A restart
+		// then resumes behind rows the segment record says are applied.
+		return i.buffer.MaybeSeal(context.Background())
 
 	case "_blocks_":
 		blockNum, ok := values[0].(uint64)
@@ -101,8 +109,10 @@ func (i *localBufferInserter) insert(table string, values []any, database *Datab
 	return i.buffer.Insert(table, values)
 }
 
+// flush has nothing to do: a spooled row is already on disk, and the segment is sealed
+// when the cursor covering it is recorded, which happens after this.
 func (i *localBufferInserter) flush(database *Database) error {
-	return i.buffer.MaybeSeal(context.Background())
+	return nil
 }
 
 func (i *localBufferInserter) close(ctx context.Context) error {
diff --git a/sink/sql/db_proto/sql/postgres/pgapplier.go b/sink/sql/db_proto/sql/postgres/pgapplier.go
index b0507417e..c0076371d 100644
--- a/sink/sql/db_proto/sql/postgres/pgapplier.go
+++ b/sink/sql/db_proto/sql/postgres/pgapplier.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	sink "github.com/streamingfast/substreams/sink"
 	"github.com/streamingfast/substreams/sink/sql/db_proto/sql/postgres/pgcopy"
 	"github.com/streamingfast/substreams/sink/sql/db_proto/sql/spool"
 	"go.uber.org/zap"
@@ -30,8 +31,9 @@ type pgApplier struct {
 	// manifest so that segments written by an earlier build still apply correctly.
 	copyRanks map[string]int
 
-	// applied is the segments table read once, on the first recovery question.
-	applied map[uint64]bool
+	// applied is the segments table read once, on the first recovery question: first block
+	// to last block of every segment the database already holds.
+	applied map[uint64]uint64
 }
 
 func newPGApplier(pool *pgxpool.Pool, schema string, copyRanks map[string]int, logger *zap.Logger) *pgApplier {
@@ -74,6 +76,46 @@ func (a *pgApplier) EnsureSchema(ctx context.Context) error {
 		return fmt.Errorf("creating the applied-segments table: %w", err)
 	}
 
+	return a.dropSegmentsPastCursor(ctx)
+}
+
+// dropSegmentsPastCursor removes the segment records the stored cursor does not cover.
+//
+// The cursor is what a restart resumes from, and Run undoes every row above it before a
+// block arrives — so a record reaching past it describes rows that are no longer there.
+// Left in place it answers AlreadyApplied for the segment carrying exactly those rows,
+// which recovery then discards while replaying the segments behind it, moving the cursor
+// over a gap it just created.
+//
+// Segments sealed before the cursor covering them was recorded are how such a record came
+// about, which no longer happens; this also clears the ones a database synced by an earlier
+// build is still carrying.
+func (a *pgApplier) dropSegmentsPastCursor(ctx context.Context) error {
+	var stored string
+	err := a.pool.QueryRow(ctx, fmt.Sprintf(`SELECT cursor FROM %s WHERE name = 'cursor'`,
+		pgx.Identifier{a.schema, "_cursor_"}.Sanitize())).Scan(&stored)
+	if err != nil {
+		// No cursor yet means nothing has been applied, so there is nothing to contradict.
+		return nil
+	}
+
+	cursor, err := sink.NewCursor(stored)
+	if err != nil || cursor.IsBlank() {
+		//nolint:nilerr // an unreadable cursor is not this pass's to report
+		return nil
+	}
+
+	tag, err := a.pool.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE last_block > $1`, a.segmentsTable()),
+		cursor.Block().Num())
+	if err != nil {
+		return fmt.Errorf("dropping the segment records past the stored cursor: %w", err)
+	}
+
+	if tag.RowsAffected() > 0 {
+		a.logger.Info("dropped segment records the stored cursor does not cover",
+			zap.Int64("dropped", tag.RowsAffected()), zap.Uint64("cursor_block", cursor.Block().Num()))
+	}
+
 	return nil
 }
 
@@ -182,26 +224,47 @@ func (a *pgApplier) AlreadyApplied(ctx context.Context, manifest *spool.Manifest
 		a.applied = applied
 	}
 
-	return a.applied[manifest.FirstBlock], nil
+	// Both ends have to match. Segments are keyed by their first block, but the block a
+	// segment ends on is whatever the sizer settled on at the time, so a range re-streamed
+	// after its segment was discarded comes back starting at the same block and ending
+	// somewhere else. Answering from the first block alone would call that one applied and
+	// drop every row it carries.
+	lastBlock, found := a.applied[manifest.FirstBlock]
+
+	return found && lastBlock == manifest.LastBlock, nil
 }
 
-// appliedSegments returns the first_block of every segment already in the database, so
+// appliedSegments returns the block range of every segment already in the database, so
 // recovery can tell what still needs replaying.
-func (a *pgApplier) appliedSegments(ctx context.Context) (map[uint64]bool, error) {
-	rows, err := a.pool.Query(ctx, fmt.Sprintf(`SELECT first_block FROM %s`, a.segmentsTable()))
+func (a *pgApplier) appliedSegments(ctx context.Context) (map[uint64]uint64, error) {
+	rows, err := a.pool.Query(ctx, fmt.Sprintf(`SELECT first_block, last_block FROM %s`, a.segmentsTable()))
 	if err != nil {
 		return nil, fmt.Errorf("listing applied segments: %w", err)
 	}
 	defer rows.Close()
 
-	applied := map[uint64]bool{}
+	applied := map[uint64]uint64{}
 	for rows.Next() {
-		var firstBlock uint64
-		if err := rows.Scan(&firstBlock); err != nil {
+		var firstBlock, lastBlock uint64
+		if err := rows.Scan(&firstBlock, &lastBlock); err != nil {
 			return nil, fmt.Errorf("scanning an applied segment: %w", err)
 		}
-		applied[firstBlock] = true
+		applied[firstBlock] = lastBlock
 	}
 
 	return applied, rows.Err()
 }
+
+// clearSegments empties the bookkeeping table.
+//
+// It is only ever called once the spool has been drained and closed, which leaves no
+// segment on disk for a record to answer for: the records exist to tell recovery whether a
+// directory it found was already applied, and there are no directories left.
+func (a *pgApplier) clearSegments(ctx context.Context) error {
+	if _, err := a.pool.Exec(ctx, fmt.Sprintf(`DELETE FROM %s`, a.segmentsTable())); err != nil {
+		return fmt.Errorf("clearing the applied-segments table: %w", err)
+	}
+	a.applied = nil
+
+	return nil
+}
diff --git a/sink/sql/db_proto/sql/postgres/pgcodec.go b/sink/sql/db_proto/sql/postgres/pgcodec.go
index a5399e3ec..feef27d38 100644
--- a/sink/sql/db_proto/sql/postgres/pgcodec.go
+++ b/sink/sql/db_proto/sql/postgres/pgcodec.go
@@ -102,7 +102,9 @@ func verifyTrailer(path string) error {
 	return nil
 }
 
-// rowWriter is one table's open stream, whichever format it is in.
+// rowWriter is one table's open stream, whichever format it is in. Closing one closes the
+// file it was writing to: the segment hands the stream its file and never touches it
+// again, so nothing else is left to release it.
 type rowWriter interface {
 	WriteRow(values []any) error
 	Rows() int64
@@ -110,6 +112,26 @@ type rowWriter interface {
 	Close() error
 }
 
+// pgCopyWriter closes the file under a binary COPY stream.
+//
+// pgcopy.Writer.Close only writes the trailer and flushes, deliberately leaving the
+// underlying writer to whoever opened it — which here is the segment, and a segment is
+// sealed and forgotten. Without this every sealed segment leaks one descriptor per table,
+// and the applier's os.RemoveAll then frees no space while they are held.
+type pgCopyWriter struct {
+	*pgcopy.Writer
+	file *os.File
+}
+
+func (w *pgCopyWriter) Close() error {
+	err := w.Writer.Close()
+	if closeErr := w.file.Close(); err == nil {
+		err = closeErr
+	}
+
+	return err
+}
+
 // pgTableFile is one table's stream within a segment. Under FormatRowLog there is no
 // per-table stream and writer stays nil: the record carries only the column layout a
 // replay needs, and the row counter.
@@ -214,7 +236,7 @@ func (s *pgSegment) openTable(table string, layout *pgcopy.Table) (*pgTableFile,
 			file.Close()
 			return nil, fmt.Errorf("starting pgcopy stream for %q: %w", table, err)
 		}
-		target.path, target.file, target.writer = path, file, writer
+		target.path, target.file, target.writer = path, file, &pgCopyWriter{Writer: writer, file: file}
 	}
 
 	s.tables[table] = target
diff --git a/sink/sql/db_proto/sql/postgres/pgcodec_close_test.go b/sink/sql/db_proto/sql/postgres/pgcodec_close_test.go
new file mode 100644
index 000000000..5bf321b77
--- /dev/null
+++ b/sink/sql/db_proto/sql/postgres/pgcodec_close_test.go
@@ -0,0 +1,49 @@
+package postgres
+
+import (
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	sqlbytes "github.com/streamingfast/substreams/sink/sql/bytes"
+	"github.com/streamingfast/substreams/sink/sql/db_proto/sql/postgres/pgcopy"
+	"github.com/streamingfast/substreams/sink/sql/db_proto/sql/spool"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSealClosesEveryStreamFile covers the descriptors a sealed segment must not keep.
+//
+// A segment is sealed and forgotten, so whatever it opened has to be released by Seal. A
+// leak here is invisible until a long backfill runs out of descriptors, and the applier's
+// os.RemoveAll frees no space while they are held.
+func TestSealClosesEveryStreamFile(t *testing.T) {
+	const tableName = "payloads"
+
+	for _, format := range []spool.Format{spool.FormatPGCopy, spool.FormatTuples} {
+		t.Run(string(format), func(t *testing.T) {
+			codec := newPGCodec(
+				format,
+				map[string]*pgcopy.Table{
+					tableName: {
+						Schema: "public",
+						Name:   tableName,
+						Columns: []pgcopy.Column{
+							{Name: "payload", OID: pgtype.TextOID},
+						},
+					},
+				},
+				sqlbytes.EncodingHex,
+			)
+			segment := &pgSegment{
+				dir:    t.TempDir(),
+				codec:  codec,
+				tables: map[string]*pgTableFile{},
+			}
+
+			require.NoError(t, segment.WriteRow(tableName, []any{[]byte{0xde, 0xad}}))
+			require.NoError(t, segment.Seal(&spool.Manifest{}))
+
+			// Closing an already closed file is what says the seal released it.
+			require.Error(t, segment.tables[tableName].file.Close())
+		})
+	}
+}
diff --git a/sink/sql/db_proto/sql/postgres/pgcopy/normalize.go b/sink/sql/db_proto/sql/postgres/pgcopy/normalize.go
index a863bc6ea..1ebb05bc8 100644
--- a/sink/sql/db_proto/sql/postgres/pgcopy/normalize.go
+++ b/sink/sql/db_proto/sql/postgres/pgcopy/normalize.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/jackc/pgx/v5/pgtype"
 	sqlbytes "github.com/streamingfast/substreams/sink/sql/bytes"
+	sql2 "github.com/streamingfast/substreams/sink/sql/db_proto/sql"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -120,9 +121,16 @@ func NormalizeRowWithEncoding(cols []Column, values []any, encoding sqlbytes.Enc
 // the pgtype array codec can then encode against the array's element OID.
 func normalizeSlice(oid uint32, in []any, encoding sqlbytes.Encoding) (any, error) {
 	if len(in) == 0 {
-		// Element type does not matter for an empty array, but the slice must still be
-		// typed for the codec to find a plan.
-		return []string{}, nil
+		// An empty array carries no element to infer a type from, and pgtype resolves the
+		// encode plan from the Go type rather than from the column: []string reaches no
+		// plan at all on a numeric[], bigint[], bytea[], bool[] or timestamp[] column, and
+		// the row is refused. The walker's own []any encodes against every array OID, so
+		// the right thing to do with it is nothing.
+		//
+		// Nothing is rendered here, unlike the element cases below: with no element there
+		// is no value whose representation could drift from what the rendered write modes
+		// produce, only a plan that has to resolve.
+		return in, nil
 	}
 
 	switch in[0].(type) {
@@ -180,6 +188,26 @@ func normalizeSlice(oid uint32, in []any, encoding sqlbytes.Encoding) (any, erro
 		}
 		return out, nil
 
+	case sql2.EnumValue:
+		// The dialect declares an enum column TEXT, so a repeated enum is TEXT[].
+		//
+		// Handing the values over as they are would encode identically today — pgtype picks
+		// the Stringer, which is what ValueToString renders with too. What it would not do
+		// is keep the two agreeing: pgtype chooses between TextValuer, driver.Valuer and
+		// fmt.Stringer in that order, so the day EnumValue gains any of the earlier ones,
+		// binary COPY starts writing something the rendered write modes do not, and nothing
+		// fails. Rendering here is what keeps that choice ours.
+		out := make([]string, len(in))
+		for i, v := range in {
+			enum, ok := v.(sql2.EnumValue)
+			if !ok {
+				return nil, fmt.Errorf("mixed element types in array: %T and sql.EnumValue", v)
+			}
+			out[i] = enum.String()
+		}
+
+		return out, nil
+
 	case int32, int64, uint32, uint64, float32, float64, bool:
 		out := make([]any, len(in))
 		for i, v := range in {
diff --git a/sink/sql/db_proto/sql/postgres/pgcopy/normalize_test.go b/sink/sql/db_proto/sql/postgres/pgcopy/normalize_test.go
index 3dbd00962..dae2ca99f 100644
--- a/sink/sql/db_proto/sql/postgres/pgcopy/normalize_test.go
+++ b/sink/sql/db_proto/sql/postgres/pgcopy/normalize_test.go
@@ -1,9 +1,11 @@
 package pgcopy
 
 import (
+	"io"
 	"testing"
 
 	"github.com/jackc/pgx/v5/pgtype"
+	sqltypes "github.com/streamingfast/substreams/sink/sql/db_proto/sql"
 	"github.com/stretchr/testify/require"
 )
 
@@ -26,6 +28,56 @@ func TestNormalizeNumericStringArrayForCopy(t *testing.T) {
 	require.Equal(t, []any{expectedLarge, expectedZero}, normalized)
 }
 
+// TestNormalizeEmptyArrayEncodesForEveryArrayType pins the empty-array shape against the
+// real encoder rather than against an expected Go type: what matters is that pgx finds a
+// binary plan for it, which is exactly what a wrongly typed empty slice fails to do.
+func TestNormalizeEmptyArrayEncodesForEveryArrayType(t *testing.T) {
+	// Every array type the from-proto dialect can declare for a repeated field.
+	oids := []uint32{
+		pgtype.NumericArrayOID,
+		pgtype.Int8ArrayOID,
+		pgtype.Int4ArrayOID,
+		pgtype.Int2ArrayOID,
+		pgtype.Float8ArrayOID,
+		pgtype.Float4ArrayOID,
+		pgtype.BoolArrayOID,
+		pgtype.ByteaArrayOID,
+		pgtype.TextArrayOID,
+		pgtype.VarcharArrayOID,
+		pgtype.TimestampArrayOID,
+		pgtype.TimestamptzArrayOID,
+		pgtype.DateArrayOID,
+		pgtype.JSONBArrayOID,
+	}
+
+	for _, oid := range oids {
+		values := []any{[]any{}}
+		columns := []Column{{Name: "values", OID: oid}}
+		require.NoError(t, NormalizeRow(columns, values), "oid %d", oid)
+
+		writer, err := NewWriter(io.Discard, columns)
+		require.NoError(t, err)
+		require.NoError(t, writer.WriteRow(values), "oid %d", oid)
+		require.NoError(t, writer.Close())
+	}
+}
+
+func TestNormalizeEnumArrayForCopy(t *testing.T) {
+	values := []any{[]any{
+		sqltypes.EnumValue{Number: 1, Name: "TRANSFER"},
+		sqltypes.EnumValue{Number: 7},
+	}}
+	columns := []Column{{Name: "kinds", OID: pgtype.TextArrayOID}}
+
+	require.NoError(t, NormalizeRow(columns, values))
+	require.Equal(t, []string{"TRANSFER", "7"}, values[0])
+
+	writer, err := NewWriter(io.Discard, columns)
+	require.NoError(t, err)
+	require.NoError(t, writer.WriteRow(values))
+	require.NoError(t, writer.Close())
+}
+
 func numericFromString(value string) (pgtype.Numeric, error) {
 	var numeric pgtype.Numeric
 	if err := numeric.Scan(value); err != nil {
diff --git a/sink/sql/db_proto/sql/postgres/types.go b/sink/sql/db_proto/sql/postgres/types.go
index 00d0e002e..36df24532 100644
--- a/sink/sql/db_proto/sql/postgres/types.go
+++ b/sink/sql/db_proto/sql/postgres/types.go
@@ -122,8 +122,15 @@ func MapFieldType(fd protoreflect.FieldDescriptor, bytesEncoding bytes.Encoding,
 	return baseType
 }
 
+// quoteLiteral renders a Go string as a PostgreSQL string literal.
+//
+// Only the quote is doubled. A backslash carries no meaning inside a standard string
+// literal, and standard_conforming_strings has been on by default since PostgreSQL 9.1, so
+// escaping it stored two backslashes where the value had one. That also made the write
+// mode visible in the data: the COPY path hands pgtype the string as it is, and only the
+// rendered paths came through here.
 func quoteLiteral(v string) string {
-	return "'" + strings.ReplaceAll(strings.ReplaceAll(v, "'", "''"), `\`, `\\`) + "'"
+	return "'" + strings.ReplaceAll(v, "'", "''") + "'"
 }
 
 func ValueToString(value any, bytesEncoding bytes.Encoding) (s string) {
@@ -182,7 +189,7 @@ func ValueToString(value any, bytesEncoding bytes.Encoding) (s string) {
 		if err != nil {
 			panic(fmt.Sprintf("failed to marshal protobuf message to JSON: %v", err))
 		}
-		s = "'" + strings.ReplaceAll(strings.ReplaceAll(string(jsonBytes), "'", "''"), "\\", "\\\\") + "'"
+		s = quoteLiteral(string(jsonBytes))
 		return
 	default:
 		if msg, ok := v.(protoreflect.ProtoMessage); ok {
@@ -190,7 +197,7 @@ func ValueToString(value any, bytesEncoding bytes.Encoding) (s string) {
 			if err != nil {
 				panic(fmt.Sprintf("failed to marshal protobuf message to JSON: %v", err))
 			}
-			s = "'" + strings.ReplaceAll(strings.ReplaceAll(string(jsonBytes), "'", "''"), "\\", "\\\\") + "'"
+			s = quoteLiteral(string(jsonBytes))
 			return
 		}
 		panic(fmt.Sprintf("unsupported type: %T", v))
diff --git a/sink/sql/db_proto/sql/postgres/types_test.go b/sink/sql/db_proto/sql/postgres/types_test.go
new file mode 100644
index 000000000..1871260a2
--- /dev/null
+++ b/sink/sql/db_proto/sql/postgres/types_test.go
@@ -0,0 +1,41 @@
+package postgres
+
+import (
+	"testing"
+
+	"github.com/streamingfast/substreams/sink/sql/bytes"
+	sql "github.com/streamingfast/substreams/sink/sql/db_proto/sql"
+	"github.com/stretchr/testify/require"
+)
+
+// TestValueToStringStoresWhatCopyWouldStore pins the rendered write modes against the
+// binary COPY one. --write-mode picks between them for throughput, so a value that
+// survives one path and not the other makes the same substream produce different rows
+// depending on a performance flag.
+//
+// The literals below are what the server stores under standard_conforming_strings = on,
+// the default since PostgreSQL 9.1, which is also what binary COPY writes verbatim.
+func TestValueToStringStoresWhatCopyWouldStore(t *testing.T) {
+	cases := []struct {
+		name     string
+		in       string
+		expected string
+	}{
+		{name: "plain", in: "hello", expected: "'hello'"},
+		{name: "a quote is doubled", in: "it's", expected: "'it''s'"},
+		{name: "a backslash is stored as one", in: `C:\path`, expected: `'C:\path'`},
+		{name: "an escape sequence is not one", in: `a\nb`, expected: `'a\nb'`},
+		{name: "both at once", in: `a\'b`, expected: `'a\''b'`},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			require.Equal(t, c.expected, ValueToString(c.in, bytes.EncodingRaw))
+		})
+	}
+}
+
+func TestValueToStringRendersAnEnumByName(t *testing.T) {
+	require.Equal(t, "'TRANSFER'", ValueToString(sql.EnumValue{Number: 1, Name: "TRANSFER"}, bytes.EncodingRaw))
+	require.Equal(t, "'7'", ValueToString(sql.EnumValue{Number: 7}, bytes.EncodingRaw))
+}
diff --git a/sink/sql/db_proto/sql/spool/frame.go b/sink/sql/db_proto/sql/spool/frame.go
index 48cfc261a..22bec9962 100644
--- a/sink/sql/db_proto/sql/spool/frame.go
+++ b/sink/sql/db_proto/sql/spool/frame.go
@@ -5,13 +5,10 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"math"
 	"os"
 )
 
-// recordSize bounds a single framed record, so a corrupt length cannot make recovery
-// allocate an arbitrary amount before the length check has a chance to reject the file.
-const recordSize = 64 << 20
-
 // FrameWriter writes length-prefixed records.
 //
 // Framing rather than lines because a rendered tuple carries SQL literals, and a text
@@ -33,6 +30,12 @@ func NewFrameWriter(file *os.File) *FrameWriter {
 // WriteRecord appends one record made of the given fields, each length-prefixed.
 func (w *FrameWriter) WriteRecord(fields ...string) error {
 	for _, field := range fields {
+		// The prefix is four bytes, so a longer field would be written with a truncated
+		// length and read back as a shorter one — a segment that looks intact and is not.
+		if int64(len(field)) > math.MaxUint32 {
+			return fmt.Errorf("a field of %d bytes cannot be framed, the length prefix holds at most %d", len(field), int64(math.MaxUint32))
+		}
+
 		binary.BigEndian.PutUint32(w.header[:], uint32(len(field)))
 		if _, err := w.writer.Write(w.header[:]); err != nil {
 			return err
@@ -63,6 +66,12 @@ type FrameReader struct {
 	file   *os.File
 	reader *bufio.Reader
 	header [4]byte
+
+	// remaining is what is left of the file, and is what bounds a record's declared
+	// length. A fixed ceiling here used to reject rows the writer had accepted, which no
+	// restart could get past: the segment verified, failed to apply, and was replayed
+	// again on the next start.
+	remaining int64
 }
 
 func OpenFrameReader(path string) (*FrameReader, error) {
@@ -71,7 +80,13 @@ func OpenFrameReader(path string) (*FrameReader, error) {
 		return nil, err
 	}
 
-	return &FrameReader{file: file, reader: bufio.NewReaderSize(file, 1<<20)}, nil
+	info, err := file.Stat()
+	if err != nil {
+		file.Close()
+		return nil, fmt.Errorf("sizing %s: %w", path, err)
+	}
+
+	return &FrameReader{file: file, reader: bufio.NewReaderSize(file, 1<<20), remaining: info.Size()}, nil
 }
 
 func (r *FrameReader) Close() error { return r.file.Close() }
@@ -82,15 +97,22 @@ func (r *FrameReader) ReadField() (string, error) {
 		return "", err
 	}
 
+	r.remaining -= int64(len(r.header))
+
+	// A record cannot be longer than what is left of the file holding it. That is what
+	// catches a corrupt length before it allocates, and it is the whole of the bound: a
+	// record the writer produced always fits, and the row it came from was already held
+	// whole in memory to be written.
 	size := binary.BigEndian.Uint32(r.header[:])
-	if size > recordSize {
-		return "", fmt.Errorf("record of %d bytes exceeds the %d byte limit", size, recordSize)
+	if int64(size) > r.remaining {
+		return "", fmt.Errorf("a record of %d bytes does not fit in the %d bytes left of the file", size, r.remaining)
 	}
 
 	field := make([]byte, size)
 	if _, err := io.ReadFull(r.reader, field); err != nil {
 		return "", fmt.Errorf("reading a %d byte record: %w", size, err)
 	}
+	r.remaining -= int64(size)
 
 	return string(field), nil
 }
diff --git a/sink/sql/db_proto/sql/spool/frame_test.go b/sink/sql/db_proto/sql/spool/frame_test.go
new file mode 100644
index 000000000..6c9446069
--- /dev/null
+++ b/sink/sql/db_proto/sql/spool/frame_test.go
@@ -0,0 +1,80 @@
+package spool
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestFrameRoundTripsARecordLargerThanAnyFixedCeiling covers the row a rendered write mode
+// can produce and a fixed read ceiling used to refuse.
+//
+// The writer accepted it, the manifest matched the bytes on disk so Verify passed, and the
+// read failed only at apply — after which recovery replayed the same segment on every
+// start and the sink could not come up at all.
+func TestFrameRoundTripsARecordLargerThanAnyFixedCeiling(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "rows.tuples")
+	file, err := os.Create(path)
+	require.NoError(t, err)
+
+	// Comfortably past the 64MiB ceiling this used to carry.
+	field := strings.Repeat("x", (64<<20)+1)
+
+	writer := NewFrameWriter(file)
+	require.NoError(t, writer.WriteRecord(field))
+	require.NoError(t, writer.Close())
+
+	reader, err := OpenFrameReader(path)
+	require.NoError(t, err)
+	defer reader.Close()
+
+	read, err := reader.ReadField()
+	require.NoError(t, err)
+	require.Equal(t, len(field), len(read))
+}
+
+// TestFrameRejectsALengthTheFileCannotHold is what replaced the fixed ceiling: a corrupt
+// prefix still cannot make the reader allocate, because no record can be longer than what
+// is left of the file carrying it.
+func TestFrameRejectsALengthTheFileCannotHold(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "rows.tuples")
+	// A prefix claiming ~4GiB in front of three bytes of payload.
+	require.NoError(t, os.WriteFile(path, []byte{0xFF, 0xFF, 0xFF, 0xF0, 'a', 'b', 'c'}, 0o600))
+
+	reader, err := OpenFrameReader(path)
+	require.NoError(t, err)
+	defer reader.Close()
+
+	_, err = reader.ReadField()
+	require.ErrorContains(t, err, "does not fit in the 3 bytes left of the file")
+}
+
+// TestFrameReadsSeveralRecordsBack pins that the running bound does not drift, the reader
+// having to subtract both the prefix and the payload of every record it hands out.
+func TestFrameReadsSeveralRecordsBack(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "rows.log")
+	file, err := os.Create(path)
+	require.NoError(t, err)
+
+	writer := NewFrameWriter(file)
+	require.NoError(t, writer.WriteRecord("customers", "1,'alpha'"))
+	require.NoError(t, writer.WriteRecord("orders", "2,'beta'"))
+	require.NoError(t, writer.Close())
+
+	reader, err := OpenFrameReader(path)
+	require.NoError(t, err)
+	defer reader.Close()
+
+	var got []string
+	for {
+		field, err := reader.ReadField()
+		if err != nil {
+			break
+		}
+		got = append(got, field)
+	}
+	require.Equal(t, []string{"customers", "1,'alpha'", "orders", "2,'beta'"}, got)
+}
diff --git a/sink/sql/db_proto/sql/spool/recover.go b/sink/sql/db_proto/sql/spool/recover.go
index f82845e14..959653ec1 100644
--- a/sink/sql/db_proto/sql/spool/recover.go
+++ b/sink/sql/db_proto/sql/spool/recover.go
@@ -88,7 +88,13 @@ func (b *Spool) recover(ctx context.Context) error {
 		}
 
 		if err := b.applier.Apply(ctx, dir, manifest); err != nil {
-			return fmt.Errorf("replaying segment %s: %w", dir, err)
+			// Left to fail, because a segment the database refuses says something is wrong
+			// with the rows or the schema, and dropping it quietly would hide that. It
+			// does mean the sink cannot start until the cause is dealt with, so the way
+			// out has to be said out loud rather than worked out from a stack trace.
+			return fmt.Errorf("replaying segment %s: %w. The sink cannot start while it is there. "+
+				"Deleting that directory drops the %d block(s) it holds, which are then streamed again from the stored cursor",
+				dir, err, manifest.BlockCount())
 		}
 		os.RemoveAll(dir)
 		replayed++
diff --git a/sink/sql/db_proto/sql/spool/sizer.go b/sink/sql/db_proto/sql/spool/sizer.go
index 08c2d336f..35f38f89e 100644
--- a/sink/sql/db_proto/sql/spool/sizer.go
+++ b/sink/sql/db_proto/sql/spool/sizer.go
@@ -33,7 +33,7 @@ type sizer struct {
 }
 
 func newSizer(target time.Duration, maxBytes int64) *sizer {
-	return &sizer{target: target, maxBytes: maxBytes, current: segmentFloorBytes}
+	return &sizer{target: target, maxBytes: maxBytes, current: min(segmentFloorBytes, maxBytes)}
 }
 
 // size reports how large a segment should grow before it is committed.
@@ -45,18 +45,27 @@ func (s *sizer) size() int64 {
 }
 
 // observe folds one measured commit back into the target.
+//
+// The size to aim for comes from the throughput the commit actually demonstrated, not
+// from scaling the current target by target/elapsed: a segment is routinely sealed well
+// short of the target — the idle seal, the drain before an undo, the seal at shutdown —
+// and such a segment applies quickly for reasons that say nothing about how fast the
+// database is. Scaling the target on that evidence doubles it while the database may in
+// fact be slow. Deriving from bytes/elapsed cannot make that mistake, and it errs on the
+// conservative side for a short segment, whose elapsed time is mostly per-segment
+// overhead and therefore understates throughput.
 func (s *sizer) observe(bytes int64, elapsed time.Duration) {
-	if elapsed <= 0 {
+	if elapsed <= 0 || bytes <= 0 {
 		return
 	}
 
-	ratio := s.target.Seconds() / elapsed.Seconds()
-	// Clamp per step so the sizer converges instead of oscillating.
-	ratio = min(max(ratio, 0.5), 2.0)
-
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	next := int64(float64(s.current) * ratio)
-	s.current = min(max(next, segmentFloorBytes), s.maxBytes)
+	next := float64(bytes) * (s.target.Seconds() / elapsed.Seconds())
+
+	// Clamp per step so the sizer converges instead of oscillating.
+	next = min(max(next, float64(s.current)*0.5), float64(s.current)*2.0)
+
+	s.current = min(max(int64(next), segmentFloorBytes), s.maxBytes)
 }
diff --git a/sink/sql/db_proto/sql/spool/sizer_test.go b/sink/sql/db_proto/sql/spool/sizer_test.go
new file mode 100644
index 000000000..0b2e8ba2c
--- /dev/null
+++ b/sink/sql/db_proto/sql/spool/sizer_test.go
@@ -0,0 +1,75 @@
+package spool
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+const mib = int64(1) << 20
+
+func TestSizerConvergesOnMeasuredThroughput(t *testing.T) {
+	const target = 3 * time.Second
+	const maxBytes = 512 * (1 << 20)
+
+	cases := []struct {
+		name     string
+		current  int64
+		bytes    int64
+		elapsed  time.Duration
+		expected int64
+	}{
+		{
+			// The one the controller exists for: 64 MiB in 3 s is exactly the target.
+			name: "a commit at target holds the size", current: 64 * mib,
+			bytes: 64 * mib, elapsed: 3 * time.Second, expected: 64 * mib,
+		},
+		{
+			// A short seal — idle, drain or shutdown — applies fast for reasons that say
+			// nothing about throughput. The size must not grow on it.
+			name: "a short segment applied fast does not grow the size", current: 64 * mib,
+			bytes: 1 * mib, elapsed: 100 * time.Millisecond, expected: 32 * mib,
+		},
+		{
+			name: "a slow database shrinks the size, one step at a time", current: 64 * mib,
+			bytes: 64 * mib, elapsed: 30 * time.Second, expected: 32 * mib,
+		},
+		{
+			name: "a fast database grows the size, one step at a time", current: 64 * mib,
+			bytes: 64 * mib, elapsed: 500 * time.Millisecond, expected: 128 * mib,
+		},
+		{
+			name: "the floor holds", current: segmentFloorBytes,
+			bytes: 1 * mib, elapsed: 30 * time.Second, expected: segmentFloorBytes,
+		},
+		{
+			name: "the ceiling holds", current: maxBytes,
+			bytes: maxBytes, elapsed: time.Millisecond, expected: maxBytes,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			sizer := newSizer(target, maxBytes)
+			sizer.current = c.current
+
+			sizer.observe(c.bytes, c.elapsed)
+
+			require.Equal(t, c.expected, sizer.size())
+		})
+	}
+}
+
+// TestSizerIgnoresAnEmptyCommit covers the segment a drain or a shutdown seals with
+// nothing in it: it carries no measurement, so it must not move the size at all.
+func TestSizerIgnoresAnEmptyCommit(t *testing.T) {
+	sizer := newSizer(3*time.Second, 512*mib)
+	sizer.current = 64 * mib
+
+	sizer.observe(0, time.Millisecond)
+	require.Equal(t, 64*mib, sizer.size())
+
+	sizer.observe(64*mib, 0)
+	require.Equal(t, 64*mib, sizer.size())
+}
diff --git a/sink/sql/db_proto/sql/spool/spool.go b/sink/sql/db_proto/sql/spool/spool.go
index 32ab4a3dc..78291c831 100644
--- a/sink/sql/db_proto/sql/spool/spool.go
+++ b/sink/sql/db_proto/sql/spool/spool.go
@@ -44,6 +44,9 @@ func (o Options) withDefaults() Options {
 	if o.SegmentMaxBytes <= 0 {
 		o.SegmentMaxBytes = 512 << 20
 	}
+	// A normal segment must fit within the total spool budget. A single row can still
+	// exceed it; that one segment is allowed through once the spool is otherwise empty.
+	o.SegmentMaxBytes = min(o.SegmentMaxBytes, o.MaxBytes)
 	if o.MaxIdle == 0 {
 		o.MaxIdle = 10 * time.Second
 	}
@@ -243,8 +246,10 @@ func (b *Spool) Seal(ctx context.Context) error {
 	b.mutex.Unlock()
 
 	// The quota is checked before the manifest is written rather than after, and against
-	// this segment's own bytes, so the spool never exceeds the budget it was given.
+	// this segment's own bytes. An individually oversized segment cannot be split without
+	// breaking row atomicity, so it is allowed once the spool is otherwise empty.
 	if err := b.awaitQuota(ctx, pending.writer.PendingBytes()); err != nil {
+		pending.writer.Discard()
 		return err
 	}
 
@@ -254,10 +259,7 @@ func (b *Spool) Seal(ctx context.Context) error {
 		return fmt.Errorf("sealing segment %s: %w", pending.dir, err)
 	}
 
-	var bytes int64
-	for _, table := range manifest.Tables {
-		bytes += table.Bytes
-	}
+	bytes := segmentBytes(manifest)
 
 	b.bytesOnDisk.Add(bytes)
 	b.blocksAhead.Add(manifest.BlockCount())
@@ -270,13 +272,28 @@ func (b *Spool) Seal(ctx context.Context) error {
 // awaitQuota blocks until the spool has room for the incoming segment.
 func (b *Spool) awaitQuota(ctx context.Context, incoming int64) error {
 	warned := false
-	for b.bytesOnDisk.Load()+incoming > b.options.MaxBytes {
+	for {
+		onDisk := b.bytesOnDisk.Load()
+		if onDisk+incoming <= b.options.MaxBytes {
+			return nil
+		}
+
+		// A single row or segment cannot be split without breaking row atomicity. Once
+		// previously queued data has drained, allow that one oversized segment through
+		// rather than waiting forever for room it can never fit into.
+		if onDisk == 0 && incoming > b.options.MaxBytes {
+			b.logger.Warn("local spool segment exceeds its disk budget, allowing it because the spool is empty",
+				zap.String("segment", humanBytes(incoming)),
+				zap.String("quota", humanBytes(b.options.MaxBytes)))
+			return nil
+		}
+
 		if err := b.pendingError(); err != nil {
 			return err
 		}
 		if !warned {
 			b.logger.Warn("local spool is full, holding the stream until the database catches up",
-				zap.String("on_disk", humanBytes(b.bytesOnDisk.Load())),
+				zap.String("on_disk", humanBytes(onDisk)),
 				zap.String("quota", humanBytes(b.options.MaxBytes)))
 			warned = true
 		}
@@ -287,8 +304,6 @@ func (b *Spool) awaitQuota(ctx context.Context, incoming int64) error {
 			return ctx.Err()
 		}
 	}
-
-	return nil
 }
 
 // startSegmentLocked opens a new segment. The caller holds mutex.
@@ -513,6 +528,15 @@ func (b *Spool) applyLoop(ctx context.Context) {
 	}
 }
 
+func segmentBytes(manifest *Manifest) int64 {
+	bytes := manifest.LogBytes
+	for _, table := range manifest.Tables {
+		bytes += table.Bytes
+	}
+
+	return bytes
+}
+
 func humanBytes(n int64) string {
 	value := float64(n)
 	for _, unit := range []string{"B", "KiB", "MiB", "GiB"} {
diff --git a/sink/sql/db_proto/sql/spool/spool_test.go b/sink/sql/db_proto/sql/spool/spool_test.go
new file mode 100644
index 000000000..5c3d1cf50
--- /dev/null
+++ b/sink/sql/db_proto/sql/spool/spool_test.go
@@ -0,0 +1,42 @@
+package spool
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func TestAwaitQuotaAllowsAnOversizedSegmentWhenSpoolIsOtherwiseEmpty(t *testing.T) {
+	spool := &Spool{
+		options: Options{MaxBytes: 10},
+		logger:  zap.NewNop(),
+	}
+
+	require.NoError(t, spool.awaitQuota(context.Background(), 11))
+}
+
+func TestSizerRespectsMaxBytesBelowSegmentFloor(t *testing.T) {
+	sizer := newSizer(time.Second, 1)
+
+	require.Equal(t, int64(1), sizer.size())
+}
+
+func TestOptionsClampSegmentMaxToSpoolMax(t *testing.T) {
+	options := (Options{MaxBytes: 10, SegmentMaxBytes: 20}).withDefaults()
+
+	require.Equal(t, int64(10), options.SegmentMaxBytes)
+}
+
+func TestSegmentBytesIncludesRowLogBytes(t *testing.T) {
+	manifest := &Manifest{
+		LogBytes: 123,
+		Tables: []TableRecord{
+			{Bytes: 7},
+		},
+	}
+
+	require.Equal(t, int64(130), segmentBytes(manifest))
+}
diff --git a/sink/sql/tests/integration/db_proto_clickhouse_fresh_database_test.go b/sink/sql/tests/integration/db_proto_clickhouse_fresh_database_test.go
new file mode 100644
index 000000000..72770d9f3
--- /dev/null
+++ b/sink/sql/tests/integration/db_proto_clickhouse_fresh_database_test.go
@@ -0,0 +1,60 @@
+package tests
+
+import (
+	"context"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/streamingfast/substreams/sink/sql/db_proto"
+	protosql "github.com/streamingfast/substreams/sink/sql/db_proto/sql"
+	pbrelations "github.com/streamingfast/substreams/sink/sql/tests/relations"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go/modules/clickhouse"
+)
+
+// TestDbProtoClickhouseSetupOnAFreshDatabase covers pointing the sink at a server that
+// does not hold the database yet, which is what `substreams sink clickhouse setup` is for
+// and the one case every other ClickHouse test skips by creating it first.
+//
+// The schema check that runs ahead of CreateDatabase used to connect to the database it
+// was about to create. ClickHouse refuses that with UNKNOWN_DATABASE, and newClient
+// retries a failed dial forever on a context of its own, so setup never came back.
+func TestDbProtoClickhouseSetupOnAFreshDatabase(t *testing.T) {
+	outputMessageDescriptor := (*pbrelations.Output)(nil).ProtoReflect().Descriptor()
+
+	var seededDatabase string
+	clickhouseDSN, _ := setupClickhouseContainer(t, func(ctx context.Context, user, password, database, dsn string, container *clickhouse.ClickHouseContainer) error {
+		seededDatabase = database
+		return nil
+	})
+
+	// Deliberately never created: that is the whole point.
+	schemaName := "fresh_database"
+	testDSN := strings.Replace(clickhouseDSN, seededDatabase, schemaName, 1)
+
+	stateFolder := t.TempDir()
+	options := db_proto.SinkerFactoryOptions{
+		UseProtoOption: true,
+		Constraints:    protosql.DisableAllConstraints(),
+		Clickhouse: db_proto.SinkerFactoryClickhouse{
+			SinkInfoFolder: stateFolder,
+			CursorFilePath: filepath.Join(stateFolder, "cursor.txt"),
+		},
+	}.Defaults()
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := db_proto.SetupDatabaseSchema(context.Background(), testDSN, schemaName,
+			defaultOutputModuleName, outputMessageDescriptor, options, logger, tracer)
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(20 * time.Second):
+		t.Fatal("setup never returned, the schema check is dialing a database that does not exist yet")
+	}
+}
diff --git a/sink/sql/tests/integration/db_proto_postgres_live_switch_test.go b/sink/sql/tests/integration/db_proto_postgres_live_switch_test.go
index 466a7e5c1..de23c76a8 100644
--- a/sink/sql/tests/integration/db_proto_postgres_live_switch_test.go
+++ b/sink/sql/tests/integration/db_proto_postgres_live_switch_test.go
@@ -159,7 +159,7 @@ func TestDbProtoPostgresSwitchToDirectInserts(t *testing.T) {
 	write(1, "customer-1")
 	require.Zero(t, countCustomers(), "a buffered write must not have reached the database yet")
 
-	require.NoError(t, database.SwitchToDirectInserts(ctx, "stream reached the chain head"))
+	require.NoError(t, database.SwitchToDirectInserts(ctx, "stream reached the chain head", true))
 	require.Equal(t, 1, countCustomers(), "the switch drains what the buffer was holding")
 
 	write(2, "customer-2")
@@ -169,3 +169,74 @@ func TestDbProtoPostgresSwitchToDirectInserts(t *testing.T) {
 	require.NoError(t, err)
 	require.Empty(t, segments, "the buffer is closed for good, nothing accumulates in it")
 }
+
+// TestDbProtoPostgresClearsSegmentRecordsAtTheChainHead covers the bookkeeping a sink that
+// stays live would otherwise accumulate forever.
+//
+// Every segment applied records a row, and nothing was ever removing them. Reaching the
+// head drains the spool and closes it for good, which leaves no directory on disk for a
+// record to answer for — so that is where they go. A run that ends instead keeps them,
+// having no next restart to grow them.
+func TestDbProtoPostgresClearsSegmentRecordsAtTheChainHead(t *testing.T) {
+	outputMessageDescriptor := (*pbrelations.Output)(nil).ProtoReflect().Descriptor()
+	postgresContainer := sharedDbChangesPostgresContainer
+	ctx := context.Background()
+
+	testSchema := "segment_records_at_head"
+	createPostgresTestSchema(t, postgresContainer.ConnectionString, testSchema)
+
+	options := db_proto.SinkerFactoryOptions{
+		UseProtoOption:  true,
+		Constraints:     protosql.DisableAllConstraints(),
+		UseTransactions: true,
+		DecodeBatchSize: 1,
+		// One byte of segment, so every write seals and records one.
+		Spool: &spool.Options{Dir: t.TempDir(), SegmentMaxBytes: 1},
+	}.Defaults()
+
+	database, err := db_proto.SetupDatabaseSchema(ctx, postgresContainer.ConnectionString, testSchema, defaultOutputModuleName, outputMessageDescriptor, options, logger, tracer)
+	require.NoError(t, err)
+	require.NoError(t, database.Open())
+	defer database.Close(ctx)
+
+	db, err := sql.Open("postgres", postgresContainer.ConnectionString)
+	require.NoError(t, err)
+	dbx := sqlx.NewDb(db, "postgres").Unsafe()
+	defer dbx.Close()
+
+	countIn := func(table string) int {
+		var out int
+		require.NoError(t, dbx.Get(&out, fmt.Sprintf(`SELECT count(*) FROM "%s"."%s"`, testSchema, table)))
+
+		return out
+	}
+
+	write := func(blockNum uint64, customerID string) {
+		t.Helper()
+
+		require.NoError(t, database.BeginTransaction())
+		require.NoError(t, database.InsertBlock(blockNum, fmt.Sprintf("%da", blockNum), fixedBaseTime))
+		require.NoError(t, database.Insert("customers", []any{blockNum, fixedBaseTime, customerID, "name"}))
+		cursor := bstream.Cursor{
+			Step:      bstream.StepNewIrreversible,
+			Block:     bstream.NewBlockRef(fmt.Sprintf("%da", blockNum), blockNum),
+			HeadBlock: bstream.NewBlockRef(fmt.Sprintf("%da", blockNum), blockNum),
+			LIB:       bstream.NewBlockRef(fmt.Sprintf("%da", blockNum), blockNum),
+		}
+		sinkCursor, err := sink.NewCursor(cursor.ToOpaque())
+		require.NoError(t, err)
+		require.NoError(t, database.StoreCursor(sinkCursor))
+		_, err = database.Flush()
+		require.NoError(t, err)
+		require.NoError(t, database.CommitTransaction())
+	}
+
+	write(1, "customer-1")
+	write(2, "customer-2")
+	require.Positive(t, countIn("_segments_"), "a spooled backfill has to be recording the segments it applies")
+
+	require.NoError(t, database.SwitchToDirectInserts(ctx, "stream reached the chain head", true))
+
+	require.Zero(t, countIn("_segments_"), "reaching the head clears what the spool recorded")
+	require.Equal(t, 2, countIn("customers"), "clearing the bookkeeping must not touch the rows")
+}
diff --git a/sink/sql/tests/integration/db_proto_postgres_segment_cursor_test.go b/sink/sql/tests/integration/db_proto_postgres_segment_cursor_test.go
new file mode 100644
index 000000000..9e2d64fd9
--- /dev/null
+++ b/sink/sql/tests/integration/db_proto_postgres_segment_cursor_test.go
@@ -0,0 +1,172 @@
+package tests
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"testing"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/jmoiron/sqlx"
+	_ "github.com/lib/pq"
+	"github.com/streamingfast/bstream"
+	"github.com/streamingfast/substreams/manifest"
+	sink "github.com/streamingfast/substreams/sink"
+	"github.com/streamingfast/substreams/sink/sql/db_proto"
+	protosql "github.com/streamingfast/substreams/sink/sql/db_proto/sql"
+	"github.com/streamingfast/substreams/sink/sql/db_proto/sql/spool"
+	pbrelations "github.com/streamingfast/substreams/sink/sql/tests/relations"
+	"github.com/stretchr/testify/require"
+)
+
+// TestDbProtoPostgresSegmentCarriesItsOwnCursor pins the one thing that makes a recorded
+// segment trustworthy: the cursor committed with it has to cover the blocks it holds.
+//
+// A segment sealed before the cursor covering it was recorded carries the previous flush's
+// instead, so it commits claiming a range its own cursor stops short of. The next run
+// resumes at that cursor and Run's undo deletes every row above it — the segment's own —
+// while the record stays behind to answer AlreadyApplied for the segment carrying exactly
+// those rows. Recovery then discards that one and replays the segments behind it, moving
+// the cursor over a gap it just created.
+func TestDbProtoPostgresSegmentCarriesItsOwnCursor(t *testing.T) {
+	outputMessageDescriptor := (*pbrelations.Output)(nil).ProtoReflect().Descriptor()
+	postgresContainer := sharedDbChangesPostgresContainer
+	ctx := context.Background()
+
+	responses := []interface{}{
+		relationsBlockData(t, "1a", "2025-01-01", entityCustomer("customer-1", "alpha")),
+		relationsBlockData(t, "2a", "2025-01-02", entityCustomer("customer-2", "beta")),
+		relationsBlockData(t, "3a", "2025-01-03", entityCustomer("customer-3", "gamma")),
+		relationsBlockData(t, "4a", "2025-01-04", entityCustomer("customer-4", "delta")),
+	}
+
+	substreamsClientConfig := setupFakeSubstreamsServer(t, responses...)
+	substreamsPackage := substreamsTestPackage(pbrelations.File_test_relations_relations_proto, outputMessageDescriptor)
+
+	baseSink, err := sink.New(
+		sink.SubstreamsModeProduction,
+		false,
+		substreamsPackage,
+		substreamsPackage.Modules.Modules[0],
+		manifest.ModuleHash{},
+		substreamsClientConfig,
+		logger,
+		tracer,
+		sink.WithBlockRange(bstream.MustParseRange("1-5", bstream.WithExclusiveEnd())),
+		sink.WithRetryBackOff(&backoff.StopBackOff{}),
+	)
+	require.NoError(t, err)
+
+	options := db_proto.SinkerFactoryOptions{
+		UseProtoOption:  true,
+		Constraints:     protosql.ConstraintPolicy{Timing: protosql.ConstraintsManual},
+		UseTransactions: true,
+		DecodeBatchSize: 1,
+		// One byte of segment: every flush seals, so each block records a segment of its
+		// own and the cursor each one carries is visible in the table.
+		Spool: &spool.Options{Dir: t.TempDir(), SegmentMaxBytes: 1},
+	}.Defaults()
+
+	testSchema := "segment_carries_its_own_cursor"
+	createPostgresTestSchema(t, postgresContainer.ConnectionString, testSchema)
+
+	dbSinker, err := db_proto.SinkerFactory(baseSink, defaultOutputModuleName, outputMessageDescriptor, options)(ctx, postgresContainer.ConnectionString, testSchema, logger, tracer)
+	require.NoError(t, err)
+
+	require.NoError(t, dbSinker.Run(ctx))
+	require.NoError(t, dbSinker.Err())
+
+	db, err := sql.Open("postgres", postgresContainer.ConnectionString)
+	require.NoError(t, err)
+	dbx := sqlx.NewDb(db, "postgres").Unsafe()
+	defer dbx.Close()
+
+	var recorded []struct {
+		FirstBlock uint64 `db:"first_block"`
+		LastBlock  uint64 `db:"last_block"`
+		Cursor     string `db:"cursor"`
+	}
+	require.NoError(t, dbx.Select(&recorded, fmt.Sprintf(
+		`SELECT first_block, last_block, cursor FROM "%s"."_segments_" ORDER BY first_block`, testSchema)))
+	require.NotEmpty(t, recorded, "the run has to have recorded at least one segment")
+
+	for _, segment := range recorded {
+		cursor, err := sink.NewCursor(segment.Cursor)
+		require.NoError(t, err)
+
+		require.GreaterOrEqualf(t, cursor.Block().Num(), segment.LastBlock,
+			"the segment covering blocks %d-%d committed with a cursor at block %d, which does not cover it",
+			segment.FirstBlock, segment.LastBlock, cursor.Block().Num())
+	}
+}
+
+// TestDbProtoPostgresDropsSegmentRecordsPastTheCursor covers the records a database synced
+// by an earlier build is still carrying, and anything else that could leave one behind.
+//
+// Run undoes every row above the stored cursor before a block arrives, so a record
+// reaching past it describes rows that are no longer there. Left in place it answers
+// AlreadyApplied for the segment carrying exactly those rows.
+func TestDbProtoPostgresDropsSegmentRecordsPastTheCursor(t *testing.T) {
+	outputMessageDescriptor := (*pbrelations.Output)(nil).ProtoReflect().Descriptor()
+	postgresContainer := sharedDbChangesPostgresContainer
+	ctx := context.Background()
+
+	testSchema := "segment_records_past_the_cursor"
+	createPostgresTestSchema(t, postgresContainer.ConnectionString, testSchema)
+
+	openSink := func(t *testing.T) {
+		t.Helper()
+
+		substreamsClientConfig := setupFakeSubstreamsServer(t,
+			relationsBlockData(t, "1a", "2025-01-01", entityCustomer("customer-1", "alpha")))
+		substreamsPackage := substreamsTestPackage(pbrelations.File_test_relations_relations_proto, outputMessageDescriptor)
+
+		baseSink, err := sink.New(
+			sink.SubstreamsModeProduction,
+			false,
+			substreamsPackage,
+			substreamsPackage.Modules.Modules[0],
+			manifest.ModuleHash{},
+			substreamsClientConfig,
+			logger,
+			tracer,
+			sink.WithBlockRange(bstream.MustParseRange("1-2", bstream.WithExclusiveEnd())),
+			sink.WithRetryBackOff(&backoff.StopBackOff{}),
+		)
+		require.NoError(t, err)
+
+		options := db_proto.SinkerFactoryOptions{
+			UseProtoOption:  true,
+			Constraints:     protosql.ConstraintPolicy{Timing: protosql.ConstraintsManual},
+			UseTransactions: true,
+			DecodeBatchSize: 1,
+			Spool:           &spool.Options{Dir: t.TempDir(), SegmentMaxBytes: 1},
+		}.Defaults()
+
+		dbSinker, err := db_proto.SinkerFactory(baseSink, defaultOutputModuleName, outputMessageDescriptor, options)(ctx, postgresContainer.ConnectionString, testSchema, logger, tracer)
+		require.NoError(t, err)
+		require.NoError(t, dbSinker.Run(ctx))
+		require.NoError(t, dbSinker.Err())
+	}
+
+	openSink(t)
+
+	db, err := sql.Open("postgres", postgresContainer.ConnectionString)
+	require.NoError(t, err)
+	dbx := sqlx.NewDb(db, "postgres").Unsafe()
+	defer dbx.Close()
+
+	// A record reaching well past anything the run reached, as an interrupted seal under
+	// the old ordering would have left.
+	_, err = dbx.Exec(fmt.Sprintf(
+		`INSERT INTO "%s"."_segments_" (first_block, last_block, cursor) VALUES (900, 999, 'stale')`, testSchema))
+	require.NoError(t, err)
+
+	// Opening the sink again is what runs the pass.
+	openSink(t)
+
+	var remaining int
+	require.NoError(t, dbx.Get(&remaining, fmt.Sprintf(
+		`SELECT count(*) FROM "%s"."_segments_" WHERE first_block = 900`, testSchema)))
+	require.Zero(t, remaining, "the record the stored cursor does not cover must be gone")
+}
diff --git a/sink/sql/tests/integration/db_proto_postgres_write_modes_test.go b/sink/sql/tests/integration/db_proto_postgres_write_modes_test.go
index c5d859efe..3e7d71369 100644
--- a/sink/sql/tests/integration/db_proto_postgres_write_modes_test.go
+++ b/sink/sql/tests/integration/db_proto_postgres_write_modes_test.go
@@ -56,7 +56,9 @@ func TestDbProtoPostgresWriteModes(t *testing.T) {
 		responses := []interface{}{
 			relationsBlockData(t, "1a", "2025-01-01", entityCustomer("customer-1", "alpha")),
 			relationsBlockData(t, "2a", "2025-01-02", entityCustomer("customer-2", "beta"), entityCustomer("customer-3", "gamma")),
-			relationsBlockData(t, "3a", "2025-01-03", entityCustomer("customer-4", "delta")),
+			// A backslash and a quote: the rendered modes build SQL literals where COPY
+			// hands pgtype the string as it is, so this is where the two would disagree.
+			relationsBlockData(t, "3a", "2025-01-03", entityCustomer("customer-4", `delta\path 'quoted'`)),
 		}
 
 		substreamsClientConfig := setupFakeSubstreamsServer(t, responses...)
@@ -102,7 +104,7 @@ func TestDbProtoPostgresWriteModes(t *testing.T) {
 		{CustomerId: "customer-1", Name: "alpha"},
 		{CustomerId: "customer-2", Name: "beta"},
 		{CustomerId: "customer-3", Name: "gamma"},
-		{CustomerId: "customer-4", Name: "delta"},
+		{CustomerId: "customer-4", Name: `delta\path 'quoted'`},
 	}
 
 	for _, test := range []struct {
diff --git a/tools/extract-proto.go b/tools/extract-proto.go
index aae2a2bb9..7ff45027d 100644
--- a/tools/extract-proto.go
+++ b/tools/extract-proto.go
@@ -34,10 +34,12 @@ var extractProtoCmd = &cobra.Command{
 
 		With --sql, the file comes annotated for 'substreams sink postgres' in from-proto
 		mode: the schema annotations are imported, every message carries a commented-out
-		table option and every field a commented-out column option, and the annotations
-		file itself is written next to it so the result parses as-is. Uncomment what the
-		schema should declare — which field is the primary key, which are unique, which
-		reference another table — and point the sink at the result:
+		table option, every field a commented-out copy of itself per column option, and the
+		annotations file itself is written next to it so the result parses as-is. Uncomment
+		the table option; for a field, a column option lives inside the field's own
+		brackets, so replace the field with the commented variant that describes it — which
+		is the primary key, which are unique, which reference another table — and point the
+		sink at the result:
 
 		    substreams tools extract-proto --sql substreams.yaml map_events
 		    substreams sink postgres setup <manifest> --proto-file-override=./map_events.proto --dsn=...
@@ -112,7 +114,7 @@ func runExtractProtoE(cmd *cobra.Command, args []string) error {
 		}
 		fmt.Printf("Wrote %s\n", schemaTarget)
 
-		fmt.Printf("\nUncomment the options that describe the schema, then:\n"+
+		fmt.Printf("\nUncomment the table options, replace a field with the commented variant that describes it, then:\n"+
 			"  substreams sink postgres setup <manifest> --proto-file-override=%s --dsn=...\n", target)
 	}
 
@@ -219,16 +221,20 @@ func renderProtoFile(file *descriptorpb.FileDescriptorProto, outputType string,
 	}
 
 	for _, enum := range file.EnumType {
-		out.WriteString("enum " + enum.GetName() + " {\n")
-		for _, value := range enum.Value {
-			fmt.Fprintf(&out, "  %s = %d;\n", value.GetName(), value.GetNumber())
-		}
-		out.WriteString("}\n\n")
+		renderEnum(&out, enum, "")
 	}
 
 	return out.String()
 }
 
+func renderEnum(out *strings.Builder, enum *descriptorpb.EnumDescriptorProto, indent string) {
+	out.WriteString(indent + "enum " + enum.GetName() + " {\n")
+	for _, value := range enum.Value {
+		fmt.Fprintf(out, "%s  %s = %d;\n", indent, value.GetName(), value.GetNumber())
+	}
+	out.WriteString(indent + "}\n\n")
+}
+
 func hasDependency(file *descriptorpb.FileDescriptorProto, name string) bool {
 	for _, dependency := range file.Dependency {
 		if dependency == name {
@@ -253,20 +259,38 @@ func renderMessage(out *strings.Builder, file *descriptorpb.FileDescriptorProto,
 	}
 
 	for _, field := range message.Field {
-		if annotate {
-			fmt.Fprintf(out, "%s  // [("+schemaProtoPackage+".field) = { primary_key: true }]  // one per message\n", indent)
-			fmt.Fprintf(out, "%s  // [("+schemaProtoPackage+".field) = { unique: true }]\n", indent)
-			fmt.Fprintf(out, "%s  // [("+schemaProtoPackage+".field) = { foreign_key: \"other_table.column\" }]\n", indent)
+		declaration := declarationOf(message, field)
+
+		// A map is walked into a table of its own rather than into a column, so none of the
+		// column options apply to the field declaring it.
+		if annotate && mapEntryOf(message, field) == nil {
+			// A field option lives inside the field's own brackets, before the semicolon,
+			// so it cannot be a comment line of its own the way the table option can.
+			// These are whole declarations: the one that describes the column replaces the
+			// plain one under it.
+			fmt.Fprintf(out, "%s  // %s [("+schemaProtoPackage+".field) = { primary_key: true }];  // replaces the line below, one per message\n", indent, declaration)
+			fmt.Fprintf(out, "%s  // %s [("+schemaProtoPackage+".field) = { unique: true }];\n", indent, declaration)
+			fmt.Fprintf(out, "%s  // %s [("+schemaProtoPackage+".field) = { foreign_key: \"other_table.column\" }];\n", indent, declaration)
 		}
 
-		fmt.Fprintf(out, "%s  %s%s %s = %d;\n", indent, labelOf(field), typeOf(field), field.GetName(), field.GetNumber())
+		fmt.Fprintf(out, "%s  %s;\n", indent, declaration)
 
 		if annotate {
 			out.WriteString("\n")
 		}
 	}
 
+	// A nested enum is declared by the message and referenced by its fields, so leaving it
+	// out writes a file naming a type nothing declares — which is the file the command
+	// tells the operator to feed back through --proto-file-override.
+	for _, enum := range message.EnumType {
+		out.WriteString("\n")
+		renderEnum(out, enum, indent+"  ")
+	}
+
 	for _, nested := range message.NestedType {
+		// The synthetic entry type of a map field. The field renders as `map<k, v>`, which
+		// declares it again, so writing it out here would be a duplicate definition.
 		if nested.GetOptions().GetMapEntry() {
 			continue
 		}
@@ -277,6 +301,56 @@ func renderMessage(out *strings.Builder, file *descriptorpb.FileDescriptorProto,
 	out.WriteString(indent + "}\n\n")
 }
 
+// declarationOf renders a field without its trailing semicolon, so the annotated form and
+// the plain one are built from the same string.
+func declarationOf(message *descriptorpb.DescriptorProto, field *descriptorpb.FieldDescriptorProto) string {
+	if entry := mapEntryOf(message, field); entry != nil {
+		// A map field is carried in the descriptor as a repeated message of a synthetic
+		// entry type. Rendering what the descriptor says — `repeated Msg.FooEntry foo` —
+		// names a type the file does not declare, and does not parse.
+		return fmt.Sprintf("map<%s, %s> %s = %d", typeOf(entryField(entry, 1)), typeOf(entryField(entry, 2)), field.GetName(), field.GetNumber())
+	}
+
+	return fmt.Sprintf("%s%s %s = %d", labelOf(field), typeOf(field), field.GetName(), field.GetNumber())
+}
+
+// mapEntryOf returns the synthetic entry type of a map field, or nil for anything else.
+func mapEntryOf(message *descriptorpb.DescriptorProto, field *descriptorpb.FieldDescriptorProto) *descriptorpb.DescriptorProto {
+	if field.GetType() != descriptorpb.FieldDescriptorProto_TYPE_MESSAGE || field.GetLabel() != descriptorpb.FieldDescriptorProto_LABEL_REPEATED {
+		return nil
+	}
+
+	name := field.GetTypeName()
+	if index := strings.LastIndex(name, "."); index >= 0 {
+		name = name[index+1:]
+	}
+
+	for _, nested := range message.NestedType {
+		if nested.GetName() != name || !nested.GetOptions().GetMapEntry() {
+			continue
+		}
+		if entryField(nested, 1) == nil || entryField(nested, 2) == nil {
+			return nil
+		}
+
+		return nested
+	}
+
+	return nil
+}
+
+// entryField picks the key (1) or the value (2) of a map entry by its field number, which
+// is what the encoding fixes, rather than by its position in the descriptor.
+func entryField(entry *descriptorpb.DescriptorProto, number int32) *descriptorpb.FieldDescriptorProto {
+	for _, field := range entry.Field {
+		if field.GetNumber() == number {
+			return field
+		}
+	}
+
+	return nil
+}
+
 func labelOf(field *descriptorpb.FieldDescriptorProto) string {
 	if field.GetLabel() == descriptorpb.FieldDescriptorProto_LABEL_REPEATED {
 		return "repeated "
diff --git a/tools/extract-proto_test.go b/tools/extract-proto_test.go
index 07f88bd0b..288d91c02 100644
--- a/tools/extract-proto_test.go
+++ b/tools/extract-proto_test.go
@@ -37,6 +37,51 @@ func TestRenderProtoFileParsesWithAnnotations(t *testing.T) {
 						Label:  descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL.Enum(),
 						Type:   descriptorpb.FieldDescriptorProto_TYPE_UINT64.Enum(),
 					},
+					{
+						// A map field, which the descriptor carries as a repeated message
+						// of a synthetic entry type.
+						Name:     proto.String("balances"),
+						Number:   proto.Int32(3),
+						Label:    descriptorpb.FieldDescriptorProto_LABEL_REPEATED.Enum(),
+						Type:     descriptorpb.FieldDescriptorProto_TYPE_MESSAGE.Enum(),
+						TypeName: proto.String(".test.output.Event.BalancesEntry"),
+					},
+					{
+						Name:     proto.String("kind"),
+						Number:   proto.Int32(4),
+						Label:    descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL.Enum(),
+						Type:     descriptorpb.FieldDescriptorProto_TYPE_ENUM.Enum(),
+						TypeName: proto.String(".test.output.Event.Kind"),
+					},
+				},
+				NestedType: []*descriptorpb.DescriptorProto{
+					{
+						Name:    proto.String("BalancesEntry"),
+						Options: &descriptorpb.MessageOptions{MapEntry: proto.Bool(true)},
+						Field: []*descriptorpb.FieldDescriptorProto{
+							{
+								Name:   proto.String("key"),
+								Number: proto.Int32(1),
+								Label:  descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL.Enum(),
+								Type:   descriptorpb.FieldDescriptorProto_TYPE_STRING.Enum(),
+							},
+							{
+								Name:   proto.String("value"),
+								Number: proto.Int32(2),
+								Label:  descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL.Enum(),
+								Type:   descriptorpb.FieldDescriptorProto_TYPE_UINT64.Enum(),
+							},
+						},
+					},
+				},
+				EnumType: []*descriptorpb.EnumDescriptorProto{
+					{
+						Name: proto.String("Kind"),
+						Value: []*descriptorpb.EnumValueDescriptorProto{
+							{Name: proto.String("KIND_UNSPECIFIED"), Number: proto.Int32(0)},
+							{Name: proto.String("KIND_TRANSFER"), Number: proto.Int32(1)},
+						},
+					},
 				},
 			},
 		},
@@ -46,7 +91,15 @@ func TestRenderProtoFileParsesWithAnnotations(t *testing.T) {
 
 	require.Contains(t, rendered, `import "`+schemaProtoPath+`"`)
 	require.Contains(t, rendered, "// option ("+schemaProtoPackage+".table)")
-	require.Contains(t, rendered, "// [("+schemaProtoPackage+".field) = { primary_key: true }]")
+	require.Contains(t, rendered, "// string id = 1 [("+schemaProtoPackage+".field) = { primary_key: true }];")
+
+	// A map field renders as a map, not as the entry type it is carried by, and the entry
+	// type is not declared a second time.
+	require.Contains(t, rendered, "map<string, uint64> balances = 3;")
+	require.NotContains(t, rendered, "BalancesEntry")
+
+	// A nested enum is declared where it is referenced from.
+	require.Contains(t, rendered, "enum Kind {")
 
 	// Uncomment what an operator would, then require the result to parse.
 	rendered = strings.Replace(rendered, `  // option (schema.table) = { name: "event" };`, `  option (schema.table) = { name: "event" };`, 1)