feat: history pruner service

blockchain/event_filter.go

@@ -10,6 +10,7 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/core/pending"
 	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/pruner"
 )
 
 var errChunkSizeReached = errors.New("chunk size reached")
@@ -26,7 +27,7 @@ type EventFilterer interface {
 }
 
 type EventFilter struct {
-	txn            db.KeyValueStore
+	database       db.KeyValueStore
 	fromBlock      uint64
 	toBlock        uint64
 	matcher        EventMatcher
@@ -44,7 +45,7 @@ const (
 )
 
 func newEventFilter(
-	txn db.KeyValueStore,
+	database db.KeyValueStore,
 	contractAddresses []felt.Address,
 	keys [][]felt.Felt,
 	fromBlock, toBlock uint64,
@@ -53,7 +54,7 @@ func newEventFilter(
 	runningFilter *core.RunningEventFilter,
 ) *EventFilter {
 	return &EventFilter{
-		txn:            txn,
+		database:       database,
 		matcher:        NewEventMatcher(contractAddresses, keys),
 		fromBlock:      fromBlock,
 		toBlock:        toBlock,
@@ -91,7 +92,7 @@ func (e *EventFilter) SetRangeEndBlockByHash(
 	filterRange EventFilterRange,
 	blockHash *felt.Felt,
 ) error {
-	header, err := core.GetBlockHeaderByHash(e.txn, blockHash)
+	header, err := core.GetBlockHeaderByHash(e.database, blockHash)
 	if err != nil {
 		return err
 	}
@@ -100,7 +101,7 @@ func (e *EventFilter) SetRangeEndBlockByHash(
 
 // SetRangeEndBlockToL1Head sets an end of the block range to latest `l1_accepted` block
 func (e *EventFilter) SetRangeEndBlockToL1Head(filterRange EventFilterRange) error {
-	l1Head, err := core.GetL1Head(e.txn)
+	l1Head, err := core.GetL1Head(e.database)
 	if err != nil {
 		return err
 	}
@@ -151,7 +152,7 @@ func (e *EventFilter) Events(
 ) ([]FilteredEvent, ContinuationToken, error) {
 	var matchedEvents []FilteredEvent
 
-	latest, err := core.GetChainHeight(e.txn)
+	latest, err := core.GetChainHeight(e.database)
 	if err != nil {
 		return nil, ContinuationToken{}, err
 	}
@@ -164,6 +165,15 @@ func (e *EventFilter) Events(
 		startBlock = cToken.fromBlock
 	}
 
+	// Reject queries whose canonical start block has been pruned. Skipped
+	// when startBlock is in the pre-confirmed range (> latest), where
+	// retention semantics don't apply.
+	if startBlock <= latest {
+		if err := pruner.RequireRetained(e.database, startBlock); err != nil {
+			return nil, ContinuationToken{}, err
+		}
+	}
+
 	// Case [canonicalBlock, canonicalBlock]
 	if e.toBlock <= latest {
 		return e.canonicalEvents(
@@ -237,13 +247,13 @@ func (e *EventFilter) canonicalEvents(
 		lastProccessedBlock = curBlock
 
 		var header *core.Header
-		header, err = core.GetBlockHeaderByNumber(e.txn, curBlock)
+		header, err = core.GetBlockHeaderByNumber(e.database, curBlock)
 		if err != nil {
 			return nil, ContinuationToken{}, err
 		}
 
 		var receipts []*core.TransactionReceipt
-		receipts, err = core.GetReceiptsByBlockNumber(e.txn, header.Number)
+		receipts, err = core.GetReceiptsByBlockNumber(e.database, header.Number)
 		if err != nil {
 			return nil, ContinuationToken{}, err
 		}

blockchain/statebackend/deprecated.go

@@ -6,6 +6,7 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/db"
 	"github.com/NethermindEth/juno/db/memory"
+	"github.com/NethermindEth/juno/pruner"
 )
 
 type deprecatedStateBackend struct {
@@ -33,15 +34,12 @@ func (b *deprecatedStateBackend) HeadState() (core.StateReader, StateCloser, err
 func (b *deprecatedStateBackend) StateAtBlockNumber(
 	blockNumber uint64,
 ) (core.StateReader, StateCloser, error) {
-	//nolint:staticcheck,nolintlint // used by old state
-	txn := b.database.NewIndexedBatch()
-
-	// Note(Ege): Why do we fetch header here? To validate block exists?
-	_, err := core.GetBlockHeaderByNumber(txn, blockNumber)
+	_, err := pruner.HeaderByNumberIfStateRetained(b.database, blockNumber)
 	if err != nil {
 		return nil, nil, err
 	}
-
+	//nolint:staticcheck,nolintlint // used by old state
+	txn := b.database.NewIndexedBatch()
 	return deprecatedstate.NewHistory(
 		deprecatedstate.New(txn),
 		blockNumber,
@@ -58,12 +56,12 @@ func (b *deprecatedStateBackend) StateAtBlockHash(
 		return deprecatedstate.New(txn), NoopStateCloser, nil
 	}
 
-	txn := b.database.NewIndexedBatch() //nolint:staticcheck // indexedBatch used by old state
-	header, err := core.GetBlockHeaderByHash(txn, blockHash)
+	header, err := pruner.HeaderByHashIfStateRetained(b.database, blockHash)
 	if err != nil {
 		return nil, nil, err
 	}
 
+	txn := b.database.NewIndexedBatch() //nolint:staticcheck // indexedBatch used by old state
 	return deprecatedstate.NewHistory(
 		deprecatedstate.New(txn),
 		header.Number,

blockchain/statebackend/statebackend.go

@@ -5,6 +5,7 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/core/state"
 	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/pruner"
 )
 
 type stateBackend struct {
@@ -33,7 +34,7 @@ func (b *stateBackend) HeadState() (core.StateReader, StateCloser, error) {
 func (b *stateBackend) StateAtBlockNumber(
 	blockNumber uint64,
 ) (core.StateReader, StateCloser, error) {
-	header, err := core.GetBlockHeaderByNumber(b.database, blockNumber)
+	header, err := pruner.HeaderByNumberIfStateRetained(b.database, blockNumber)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -56,12 +57,16 @@ func (b *stateBackend) StateAtBlockHash(
 		return st, NoopStateCloser, nil
 	}
 
-	blockNumber, err := core.GetBlockHeaderNumberByHash(b.database, blockHash)
+	header, err := pruner.HeaderByHashIfStateRetained(b.database, blockHash)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	return b.StateAtBlockNumber(blockNumber)
+	history, err := state.NewStateHistory(header.Number, header.GlobalStateRoot, b.stateDB)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &history, NoopStateCloser, nil
 }
 
 func (b *stateBackend) Store(

cmd/juno/juno.go

@@ -107,6 +107,8 @@ const (
 	rpcRequestTimeoutF                  = "rpc-request-timeout"
 	maxConcurrentCompilationsF          = "max-concurrent-compilations"
 	disableReceivedTxnStreamF           = "disable-received-txn-stream"
+	newStateF                           = "new-state"
+	pruneModeF                          = node.PruneModeFlag
 
 	defaultConfig                             = ""
 	defaultLogJSON                            = false
@@ -166,7 +168,7 @@ const (
 	defaultRPCRequestTimeout                  = 1 * time.Minute
 	defaultMaxConcurrentCompilations          = 8
 	defaultDisableReceivedTxnStream           = false
-	newStateF                                 = "new-state"
+	defaultPruneMode                          = uint64(0)
 
 	configFlagUsage                       = "The YAML configuration file."
 	logLevelFlagUsage                     = "Options: trace, debug, info, warn, error."
@@ -248,7 +250,22 @@ const (
 		"Use zstd for low storage."
 	rpcRequestTimeoutUsage         = "Maximum time for an RPC request to complete."
 	maxConcurrentCompilationsUsage = "Maximum concurrent Sierra compilations."
-	disableReceivedTxnStreamUsage  = "The starknet_subscribeNewTransactions WebSocket API " +
+	pruneModeUsage                 = "Enables block-data and state-history pruning. Pruning is " +
+		"disabled by default; passing this flag (with or without a value) turns " +
+		"it on. The value is the size of the retention window in blocks, counted " +
+		"back from the latest L1-verified head:\n" +
+		"  --prune-mode      same as --prune-mode=0; prune up to the L1 head\n" +
+		"  --prune-mode=N    keep blocks in (l1_head - N, l2_head], prune below\n" +
+		"Blocks at or above the L2 head are always kept. The floor is anchored on " +
+		"the L1-verified head — never on the local L2 head — so pruned blocks are " +
+		"reorg-safe. RPC remains fully functional for any block inside the " +
+		"retention window; requests targeting blocks below the floor fail because " +
+		"their data has been deleted. Pruning is irreversible: data deleted under " +
+		"a small window cannot be recovered without re-syncing. Changing this " +
+		"value across restarts is safe: the window grows or shrinks accordingly. " +
+		"Growth is gradual — pruning pauses until the L1 head advances enough to " +
+		"reach the new floor."
+	disableReceivedTxnStreamUsage = "The starknet_subscribeNewTransactions WebSocket API " +
 		"allows users to subscribe to new transactions. By default, it streams " +
 		"transactions that have been accepted on L2. Users can optionally provide " +
 		"a set of finality statuses to be notified about, including transactions " +
@@ -359,6 +376,10 @@ func NewCmd(config *node.Config, run func(*cobra.Command, []string) error) *cobr
 			return err
 		}
 
+		// Pruning is gated on the flag's *presence* (CLI, YAML, or env), not
+		// its numeric value — --prune-mode=0 is still "on, retain 0".
+		config.Prune = v.IsSet(pruneModeF)
+
 		// Set custom network
 		if v.IsSet(cnNameF) {
 			l1ChainID, ok := new(big.Int).SetString(v.GetString(cnL1ChainIDF), 0)
@@ -489,6 +510,9 @@ func NewCmd(config *node.Config, run func(*cobra.Command, []string) error) *cobr
 	junoCmd.Flags().Bool(
 		disableReceivedTxnStreamF, defaultDisableReceivedTxnStream, disableReceivedTxnStreamUsage,
 	)
+	junoCmd.Flags().Uint64(pruneModeF, defaultPruneMode, pruneModeUsage)
+	// NoOptDefVal lets users pass --prune-mode without a value (treated as 0).
+	junoCmd.Flags().Lookup(pruneModeF).NoOptDefVal = "0"
 	junoCmd.AddCommand(GenP2PKeyPair(), DBCmd(defaultDBPath), CompileSierraCmd())
 
 	return junoCmd

core/block.go

@@ -15,6 +15,17 @@ import (
 
 type BlockSignFunc func(blockHash, stateDiffCommitment *felt.Felt) ([]*felt.Felt, error)
 
+// BlockHashLag is the Starknet protocol-defined lag for the BLOCK_HASH
+// syscall: when executing block N, contracts can read the hash of block
+// N - BlockHashLag.
+const BlockHashLag uint64 = 10
+
+// Block hash storage contract introduced with starknet v0.12.0.
+// Contract stores block number to block hash mapping,
+// and serve this information through [get_block_hash_syscall].
+// Queriable range is [first_v0_12_0_block, current_block - 10]
+var BlockHashStorageContract = &felt.One
+
 type L1Head struct {
 	BlockNumber uint64
 	BlockHash   *felt.Felt

core/typed_buckets.go

@@ -164,7 +164,7 @@ var BlockTransactionsBucket = prefix.NewPrefixedBucket(
 		key.Cbor[uint64](),
 		BlockTransactionsSerializer{},
 	),
-	prefix.Prefix(key.Uint64, prefix.End[BlockTransactions]()),
+	prefix.Prefix(key.Cbor[uint64](), prefix.End[BlockTransactions]()),
 )
 
 var BlockTransactionsTransactionPartialBucket = partial.NewPartialBucket(

node/metrics.go

@@ -13,16 +13,18 @@ import (
 	"github.com/NethermindEth/juno/jemalloc"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/l1"
+	"github.com/NethermindEth/juno/pruner"
 	"github.com/NethermindEth/juno/sync"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
 const (
 	l1MetricsTimeout = 5 * time.Second
 
-	labelMethod   = "method"
-	labelVersion  = "version"
-	namespaceSync = "sync"
+	labelMethod     = "method"
+	labelVersion    = "version"
+	namespaceSync   = "sync"
+	namespacePruner = "pruner"
 )
 
 func makeDBMetrics() db.EventListener {
@@ -360,3 +362,44 @@ func makeVMThrottlerMetrics(throttledVM *ThrottledVM) {
 	})
 	prometheus.MustRegister(vmJobs, vmQueue)
 }
+
+func makePrunerMetrics() pruner.EventListener {
+	oldestBlockKept := prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespacePruner,
+		Name:      "oldest_block_kept",
+		Help:      "Block number of the oldest block retained after pruning",
+	})
+	pruneLatency := prometheus.NewSummary(prometheus.SummaryOpts{
+		Namespace: namespacePruner,
+		Name:      "prune_latency",
+		Help:      "Time taken per prune operation in seconds",
+	})
+	blocksPruned := prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: namespacePruner,
+		Name:      "blocks_pruned_total",
+		Help:      "Total number of blocks pruned",
+	})
+	errors := prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: namespacePruner,
+		Name:      "errors_total",
+		Help:      "Total number of errors encountered while pruning",
+	})
+	lastRunTimestamp := prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespacePruner,
+		Name:      "last_run_timestamp_seconds",
+		Help:      "Unix timestamp of the last completed prune operation",
+	})
+	prometheus.MustRegister(oldestBlockKept, pruneLatency, blocksPruned, errors, lastRunTimestamp)
+
+	return &pruner.SelectiveListener{
+		OnPruneCb: func(oldest uint64, count uint64, took time.Duration) {
+			oldestBlockKept.Set(float64(oldest))
+			blocksPruned.Add(float64(count))
+			pruneLatency.Observe(took.Seconds())
+			lastRunTimestamp.SetToCurrentTime()
+		},
+		OnPruneErrorCb: func(err error) {
+			errors.Inc()
+		},
+	}
+}