feat(migration): statehistory migration

migration/statehistory/class_hash_ingestor.go

@@ -0,0 +1,162 @@
+package statehistory
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/NethermindEth/juno/core/felt"
+	"github.com/NethermindEth/juno/core/state"
+	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/db/dbutils"
+	"github.com/NethermindEth/juno/migration/pipeline"
+	"github.com/NethermindEth/juno/migration/semaphore"
+)
+
+type classHashIngestor struct {
+	baseIngestor
+}
+
+var _ pipeline.State[*felt.Felt, task] = (*classHashIngestor)(nil)
+
+func newClassHashIngestor(
+	ctx context.Context,
+	sem semaphore.ResourceSemaphore[db.Batch],
+	database db.KeyValueReader,
+) *classHashIngestor {
+	return &classHashIngestor{baseIngestor: newBaseIngestor(ctx, sem, database)}
+}
+
+// Run migrates the class-hash history of a single contract.
+//
+// Legend: Bₙ = block at which the n-th class-hash *replacement* happened.
+// Vₙ = the class hash active *after* Bₙ; V₀ is the deploy-time hash. The
+// deprecated layout writes nothing at deploy: each entry is written only
+// on a *Replace*, and the value stored is the hash that was active before
+// that replace. So deprecated[B₁] = V₀ even though no replace happened at
+// deploy_h itself. The new layout adds an explicit deploy entry and shifts
+// everything else by one slot:
+//
+//	block    │ deprecated     │ new
+//	─────────┼────────────────┼──────
+//	deploy_h │  —             │ V₀     ← inserted from first deprecated entry
+//	  B₁     │  V₀            │ V₁
+//	  B₂     │  V₁            │ V₂
+//	  B₃     │  V₂            │ V₃
+//	─────────┼────────────────┼──────
+//	  > B₃   │  contract      │ V₃ (last entry — self-contained)
+//	            .ClassHash      ← deprecated must reach into the Contract
+//	                              record for any block past the last replace
+//
+// If the deprecated history is empty (no replaces ever), the single deploy
+// entry is written with contract.ClassHash directly. Deprecated rows are
+// deleted at the end of the run. Resume-safe: empty-deprecated + existing
+// deploy entry → no-op.
+func (i *classHashIngestor) Run(index int, addr *felt.Felt, outputs chan<- task) error {
+	t := &i.tasks[index]
+
+	deprecatedPrefix := db.DeprecatedContractClassHashHistoryKey(addr)
+	contract, err := state.GetContract(i.database, addr)
+	if err != nil {
+		return fmt.Errorf("class-hash: GetContract(%s): %w", addr, err)
+	}
+
+	depIt, err := i.database.NewIterator(deprecatedPrefix, true)
+	if err != nil {
+		return fmt.Errorf("class-hash: open deprecated iter(%s): %w", addr, err)
+	}
+	defer depIt.Close()
+
+	if !depIt.First() {
+		return i.writeDeployOnly(t, outputs, addr, contract.DeployedHeight, &contract.ClassHash)
+	}
+	return i.writeShiftedHistory(
+		t, outputs, depIt, deprecatedPrefix, addr,
+		contract.DeployedHeight, &contract.ClassHash,
+	)
+}
+
+// writeDeployOnly handles the "no deprecated history" branch: write the
+// deploy-time entry from contract.ClassHash, unless a previous run already
+// wrote it.
+func (i *classHashIngestor) writeDeployOnly(
+	t *task,
+	outputs chan<- task,
+	addr *felt.Felt,
+	deployHeight uint64,
+	classHash *felt.Felt,
+) error {
+	deployKey := db.ContractClassHashHistoryAtBlockKey(addr, deployHeight)
+	deployEntryExists, err := i.database.Has(deployKey)
+	if err != nil {
+		return fmt.Errorf("class-hash: Has(deploy entry): %w", err)
+	}
+	if deployEntryExists {
+		return nil
+	}
+	if err := state.WriteClassHashHistory(t.batch, addr, deployHeight, classHash); err != nil {
+		return err
+	}
+	t.completedAddrs++
+	t.entryCount++
+	return i.flush(t, outputs)
+}
+
+// writeShiftedHistory handles the "non-empty deprecated history" branch:
+// writes the deploy entry from the first deprecated value, shifts each
+// deprecated entry into the new layout using the next entry's pre-value
+// (or contract.ClassHash for the last), and deletes the deprecated rows.
+// depIt must be positioned at the first deprecated entry.
+func (i *classHashIngestor) writeShiftedHistory(
+	t *task,
+	outputs chan<- task,
+	depIt db.Iterator,
+	prefix []byte,
+	addr *felt.Felt,
+	deployHeight uint64,
+	headClassHash *felt.Felt,
+) error {
+	rawValue, err := depIt.Value()
+	if err != nil {
+		return fmt.Errorf("class-hash: read first value(%s): %w", addr, err)
+	}
+	deployClassHash := felt.FromBytes[felt.Felt](rawValue)
+	if err := state.WriteClassHashHistory(t.batch, addr, deployHeight, &deployClassHash); err != nil {
+		return err
+	}
+	t.entryCount++
+	if err := i.flush(t, outputs); err != nil {
+		return err
+	}
+
+	for {
+		block, err := parseBlockKey(depIt.Key(), prefix)
+		if err != nil {
+			return fmt.Errorf("class-hash(%s): %w", addr, err)
+		}
+		hasNext := depIt.Next()
+		historyValue := *headClassHash
+		if hasNext {
+			rawValue, err := depIt.Value()
+			if err != nil {
+				return fmt.Errorf("class-hash(%s): %w", addr, err)
+			}
+			historyValue = felt.FromBytes[felt.Felt](rawValue)
+		}
+		if err := state.WriteClassHashHistory(t.batch, addr, block, &historyValue); err != nil {
+			return err
+		}
+		t.entryCount++
+		if err := i.flush(t, outputs); err != nil {
+			return err
+		}
+		if !hasNext {
+			break
+		}
+	}
+
+	if err := t.batch.DeleteRange(prefix, dbutils.UpperBound(prefix)); err != nil {
+		return fmt.Errorf("class-hash: DeleteRange deprecated(%s): %w", addr, err)
+	}
+	t.completedAddrs++
+	return nil
+}

migration/statehistory/committer.go

@@ -0,0 +1,52 @@
+package statehistory
+
+import (
+	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/migration/pipeline"
+	"github.com/NethermindEth/juno/migration/semaphore"
+	"github.com/NethermindEth/juno/utils/log"
+	"go.uber.org/zap"
+)
+
+type committer struct {
+	logger         log.StructuredLogger
+	counter        counter
+	batchSemaphore semaphore.ResourceSemaphore[db.Batch]
+}
+
+var _ pipeline.State[task, struct{}] = (*committer)(nil)
+
+func newCommitter(
+	logger log.StructuredLogger,
+	batchSemaphore semaphore.ResourceSemaphore[db.Batch],
+	phaseName string,
+) *committer {
+	return &committer{
+		logger:         logger,
+		counter:        newCounter(logger, timeLogRate, phaseName),
+		batchSemaphore: batchSemaphore,
+	}
+}
+
+func (c *committer) Run(_ int, t task, _ chan<- struct{}) error {
+	defer c.batchSemaphore.Put()
+
+	c.logger.Debug(
+		"writing batch",
+		zap.Int("completedAddrs", t.completedAddrs),
+		zap.Int("entryCount", t.entryCount),
+		zap.Int("batchSize", t.batch.Size()),
+	)
+
+	byteSize := uint64(t.batch.Size())
+	if err := t.batch.Write(); err != nil {
+		return err
+	}
+
+	c.counter.log(byteSize, t.completedAddrs, t.entryCount)
+	return nil
+}
+
+func (c *committer) Done(int, chan<- struct{}) error {
+	return nil
+}

migration/statehistory/counter.go

@@ -0,0 +1,55 @@
+package statehistory
+
+import (
+	"time"
+
+	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/utils/log"
+	"go.uber.org/zap"
+)
+
+type counter struct {
+	logger         log.StructuredLogger
+	timeLogRate    time.Duration
+	phaseName      string
+	start          time.Time
+	size           uint64
+	completedAddrs uint64
+	entryCount     uint64
+}
+
+func newCounter(logger log.StructuredLogger, timeLogRate time.Duration, phaseName string) counter {
+	return counter{
+		logger:      logger,
+		timeLogRate: timeLogRate,
+		phaseName:   phaseName,
+		start:       time.Now(),
+	}
+}
+
+func (c *counter) log(byteSize uint64, completedAddrs, entryCount int) {
+	c.size += byteSize
+	c.completedAddrs += uint64(completedAddrs)
+	c.entryCount += uint64(entryCount)
+
+	now := time.Now()
+	elapsed := now.Sub(c.start).Seconds()
+	if elapsed > c.timeLogRate.Seconds() {
+		mbs := float64(c.size) / float64(db.Megabyte)
+		c.logger.Info(
+			"write speed",
+			zap.String("phase", c.phaseName),
+			zap.Float64("MB", mbs),
+			zap.Float64("MB/s", mbs/elapsed),
+			zap.Uint64("completedContracts", c.completedAddrs),
+			zap.Float64("completedContracts/s", float64(c.completedAddrs)/elapsed),
+			zap.Uint64("entries", c.entryCount),
+			zap.Float64("entries/s", float64(c.entryCount)/elapsed),
+			zap.Float64("time", elapsed),
+		)
+		c.start = now
+		c.size = 0
+		c.completedAddrs = 0
+		c.entryCount = 0
+	}
+}

migration/statehistory/ingestor.go

@@ -0,0 +1,63 @@
+package statehistory
+
+import (
+	"context"
+
+	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/migration/semaphore"
+)
+
+type baseIngestor struct {
+	ctx            context.Context
+	batchSemaphore semaphore.ResourceSemaphore[db.Batch]
+	database       db.KeyValueReader
+	tasks          []task
+}
+
+// newBaseIngestor pre-allocates one batch per ingestor slot. The semaphore is
+// created with capacity ingestorCount+1 immediately before this call, so the
+// acquires cannot block — using GetBlocking keeps the constructor signature
+// error-free.
+func newBaseIngestor(
+	ctx context.Context,
+	sem semaphore.ResourceSemaphore[db.Batch],
+	database db.KeyValueReader,
+) baseIngestor {
+	tasks := make([]task, ingestorCount)
+	for i := range tasks {
+		tasks[i] = task{batch: sem.GetBlocking()}
+	}
+	return baseIngestor{
+		ctx:            ctx,
+		batchSemaphore: sem,
+		database:       database,
+		tasks:          tasks,
+	}
+}
+
+// flush emits the current task downstream when its batch hits target size and
+// acquires a fresh batch. The ctx-aware select on the channel send is the
+// snappy cancellation point. The semaphore acquire uses GetBlocking — it is
+// guaranteed to unblock within one committer iteration because the committer's
+// deferred Put always runs.
+func (b *baseIngestor) flush(t *task, outputs chan<- task) error {
+	if t.batch.Size() < targetBatchByteSize {
+		return nil
+	}
+	select {
+	case <-b.ctx.Done():
+		return b.ctx.Err()
+	case outputs <- *t:
+	}
+	*t = task{batch: b.batchSemaphore.GetBlocking()}
+	return nil
+}
+
+func (b *baseIngestor) Done(index int, outputs chan<- task) error {
+	select {
+	case <-b.ctx.Done():
+		return b.ctx.Err()
+	case outputs <- b.tasks[index]:
+	}
+	return nil
+}