diff --git a/block/internal/executing/executor.go b/block/internal/executing/executor.go index 845e18727..d2efca0f7 100644 --- a/block/internal/executing/executor.go +++ b/block/internal/executing/executor.go @@ -430,6 +430,23 @@ func (e *Executor) produceBlock() error { // Update in-memory state after successful commit e.setLastState(newState) + // Run height-based pruning of stored block data if enabled. This is a + // best-effort background maintenance step and should not cause block + // production to fail, but it does run in the critical path and may add + // some latency when large ranges are pruned. + if e.config.Node.PruningEnabled && e.config.Node.PruningKeepRecent > 0 && e.config.Node.PruningInterval > 0 { + if newHeight%e.config.Node.PruningInterval == 0 { + // Compute the prune floor: all heights <= targetHeight are candidates + // for pruning of header/data/signature/index entries. + if newHeight > e.config.Node.PruningKeepRecent { + targetHeight := newHeight - e.config.Node.PruningKeepRecent + if err := e.store.PruneBlocks(e.ctx, targetHeight); err != nil { + e.logger.Error().Err(err).Uint64("target_height", targetHeight).Msg("failed to prune old block data") + } + } + } + } + // broadcast header and data to P2P network g, ctx := errgroup.WithContext(e.ctx) g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, header) }) diff --git a/pkg/config/config.go b/pkg/config/config.go index 6f358e876..ca7b352c0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -210,6 +210,13 @@ type NodeConfig struct { // Readiness / health configuration ReadinessWindowSeconds uint64 `mapstructure:"readiness_window_seconds" yaml:"readiness_window_seconds" comment:"Time window in seconds used to calculate ReadinessMaxBlocksBehind based on block time. Default: 15 seconds."` ReadinessMaxBlocksBehind uint64 `mapstructure:"readiness_max_blocks_behind" yaml:"readiness_max_blocks_behind" comment:"How many blocks behind best-known head the node can be and still be considered ready. 0 means must be exactly at head."` + + // Pruning configuration + // When enabled, the node will periodically prune old block data (headers, data, + // signatures, and hash index) from the local store while keeping recent history. + PruningEnabled bool `mapstructure:"pruning_enabled" yaml:"pruning_enabled" comment:"Enable height-based pruning of stored block data. When disabled, all blocks are kept (archive mode)."` + PruningKeepRecent uint64 `mapstructure:"pruning_keep_recent" yaml:"pruning_keep_recent" comment:"Number of most recent blocks to retain. Older blocks will have their header/data/signature removed from the local store. 0 means keep all blocks."` + PruningInterval uint64 `mapstructure:"pruning_interval" yaml:"pruning_interval" comment:"Run pruning every N blocks. Must be >= 1 when pruning is enabled."` } // LogConfig contains all logging configuration parameters @@ -341,6 +348,10 @@ func AddFlags(cmd *cobra.Command) { cmd.Flags().Uint64(FlagReadinessWindowSeconds, def.Node.ReadinessWindowSeconds, "time window in seconds for calculating readiness threshold based on block time (default: 15s)") cmd.Flags().Uint64(FlagReadinessMaxBlocksBehind, def.Node.ReadinessMaxBlocksBehind, "how many blocks behind best-known head the node can be and still be considered ready (0 = must be at head)") cmd.Flags().Duration(FlagScrapeInterval, def.Node.ScrapeInterval.Duration, "interval at which the reaper polls the execution layer for new transactions") + // Pruning configuration flags + cmd.Flags().Bool(FlagPrefixEvnode+"node.pruning_enabled", def.Node.PruningEnabled, "enable height-based pruning of stored block data (headers, data, signatures, index)") + cmd.Flags().Uint64(FlagPrefixEvnode+"node.pruning_keep_recent", def.Node.PruningKeepRecent, "number of most recent blocks to retain when pruning is enabled (0 = keep all)") + cmd.Flags().Uint64(FlagPrefixEvnode+"node.pruning_interval", def.Node.PruningInterval, "run pruning every N blocks (must be >= 1 when pruning is enabled)") // Data Availability configuration flags cmd.Flags().String(FlagDAAddress, def.DA.Address, "DA address (host:port)") diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 062c9fe19..8ce28be2b 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -69,6 +69,9 @@ func DefaultConfig() Config { ReadinessWindowSeconds: defaultReadinessWindowSeconds, ReadinessMaxBlocksBehind: calculateReadinessMaxBlocksBehind(defaultBlockTime.Duration, defaultReadinessWindowSeconds), ScrapeInterval: DurationWrapper{1 * time.Second}, + PruningEnabled: false, + PruningKeepRecent: 0, + PruningInterval: 0, }, DA: DAConfig{ Address: "http://localhost:7980", diff --git a/pkg/store/keys.go b/pkg/store/keys.go index dd989c0e8..ff96fd895 100644 --- a/pkg/store/keys.go +++ b/pkg/store/keys.go @@ -25,6 +25,12 @@ const ( // LastSubmittedHeaderHeightKey is the key used for persisting the last submitted header height in store. LastSubmittedHeaderHeightKey = "last-submitted-header-height" + // LastPrunedBlockHeightKey is the metadata key used for persisting the last + // pruned block height in the store. All block data (header, data, + // signature, and hash index) for heights <= this value are considered + // pruned and may be missing from the store. + LastPrunedBlockHeightKey = "last-pruned-block-height" + headerPrefix = "h" dataPrefix = "d" signaturePrefix = "c" diff --git a/pkg/store/store.go b/pkg/store/store.go index 972b94e0e..782ff2821 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -274,6 +274,85 @@ func (s *DefaultStore) Rollback(ctx context.Context, height uint64, aggregator b return nil } +// PruneBlocks removes block data (header, data, signature, and hash index) +// up to and including the given height from the store. It does not modify +// the current chain height or any state snapshots. +// +// This method is intended for long-term storage reduction and is safe to +// call repeatedly with the same or increasing heights. +func (s *DefaultStore) PruneBlocks(ctx context.Context, height uint64) error { + batch, err := s.db.Batch(ctx) + if err != nil { + return fmt.Errorf("failed to create a new batch for pruning: %w", err) + } + + // Track the last successfully pruned height so we can resume across restarts. + var lastPruned uint64 + meta, err := s.GetMetadata(ctx, LastPrunedBlockHeightKey) + if err != nil { + if !errors.Is(err, ds.ErrNotFound) { + return fmt.Errorf("failed to get last pruned height: %w", err) + } + } else if len(meta) == heightLength { + lastPruned, err = decodeHeight(meta) + if err != nil { + return fmt.Errorf("failed to decode last pruned height: %w", err) + } + } + + // Nothing new to prune. + if height <= lastPruned { + return nil + } + + // Delete block data for heights in (lastPruned, height]. + for h := lastPruned + 1; h <= height; h++ { + // Get header blob to compute the hash index key. If header is already + // missing (e.g. due to previous partial pruning), just skip this height. + headerBlob, err := s.db.Get(ctx, ds.NewKey(getHeaderKey(h))) + if err != nil { + if errors.Is(err, ds.ErrNotFound) { + continue + } + return fmt.Errorf("failed to get header at height %d during pruning: %w", h, err) + } + + if err := batch.Delete(ctx, ds.NewKey(getHeaderKey(h))); err != nil { + return fmt.Errorf("failed to delete header at height %d during pruning: %w", h, err) + } + + if err := batch.Delete(ctx, ds.NewKey(getDataKey(h))); err != nil { + if !errors.Is(err, ds.ErrNotFound) { + return fmt.Errorf("failed to delete data at height %d during pruning: %w", h, err) + } + } + + if err := batch.Delete(ctx, ds.NewKey(getSignatureKey(h))); err != nil { + if !errors.Is(err, ds.ErrNotFound) { + return fmt.Errorf("failed to delete signature at height %d during pruning: %w", h, err) + } + } + + headerHash := sha256.Sum256(headerBlob) + if err := batch.Delete(ctx, ds.NewKey(getIndexKey(headerHash[:]))); err != nil { + if !errors.Is(err, ds.ErrNotFound) { + return fmt.Errorf("failed to delete index for height %d during pruning: %w", h, err) + } + } + } + + // Persist the updated last pruned height. + if err := batch.Put(ctx, ds.NewKey(getMetaKey(LastPrunedBlockHeightKey)), encodeHeight(height)); err != nil { + return fmt.Errorf("failed to update last pruned height: %w", err) + } + + if err := batch.Commit(ctx); err != nil { + return fmt.Errorf("failed to commit pruning batch: %w", err) + } + + return nil +} + const heightLength = 8 func encodeHeight(height uint64) []byte { diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go index 8636a0ad0..b9a796788 100644 --- a/pkg/store/store_test.go +++ b/pkg/store/store_test.go @@ -734,6 +734,57 @@ func TestRollback(t *testing.T) { require.Equal(rollbackToHeight, state.LastBlockHeight) } +func TestPruneBlocks_RemovesOldBlockDataOnly(t *testing.T) { + t.Parallel() + + ctx := context.Background() + ds, err := NewTestInMemoryKVStore() + require.NoError(t, err) + + s := New(ds).(*DefaultStore) + + // create and store a few blocks with headers, data, signatures and state + batch, err := s.NewBatch(ctx) + require.NoError(t, err) + + var lastState types.State + for h := uint64(1); h <= 5; h++ { + header := &types.SignedHeader{Header: types.Header{BaseHeader: types.BaseHeader{Height: h}}} + data := &types.Data{} + sig := types.Signature([]byte{byte(h)}) + + require.NoError(t, batch.SaveBlockData(header, data, &sig)) + + // fake state snapshot per height + lastState = types.State{LastBlockHeight: h} + require.NoError(t, batch.UpdateState(lastState)) + } + require.NoError(t, batch.SetHeight(5)) + require.NoError(t, batch.Commit()) + + // prune everything up to height 3 + require.NoError(t, s.PruneBlocks(ctx, 3)) + + // old block data should be gone + for h := uint64(1); h <= 3; h++ { + _, _, err := s.GetBlockData(ctx, h) + assert.Error(t, err, "expected block data at height %d to be pruned", h) + } + + // recent block data should remain + for h := uint64(4); h <= 5; h++ { + _, _, err := s.GetBlockData(ctx, h) + assert.NoError(t, err, "expected block data at height %d to be kept", h) + } + + // state snapshots are not pruned by PruneBlocks + for h := uint64(1); h <= 5; h++ { + st, err := s.GetStateAtHeight(ctx, h) + assert.NoError(t, err, "expected state at height %d to remain", h) + assert.Equal(t, h, st.LastBlockHeight) + } +} + // TestRollbackToSameHeight verifies that rollback to same height is a no-op func TestRollbackToSameHeight(t *testing.T) { t.Parallel() diff --git a/pkg/store/types.go b/pkg/store/types.go index bf1cb6ced..2281e1a4c 100644 --- a/pkg/store/types.go +++ b/pkg/store/types.go @@ -30,9 +30,15 @@ type Batch interface { } // Store is minimal interface for storing and retrieving blocks, commits and state. +// +// It is composed from three concerns: +// - Reader: read access to blocks, state, and metadata +// - Rollback: consensus rollback logic (used for chain reorgs / recovery) +// - Pruner: long-term height-based pruning of historical block data type Store interface { - Rollback Reader + Rollback + Pruner // SetMetadata saves arbitrary value in the store. // @@ -75,3 +81,14 @@ type Rollback interface { // Aggregator is used to determine if the rollback is performed on the aggregator node. Rollback(ctx context.Context, height uint64, aggregator bool) error } + +// Pruner provides long-term, height-based pruning of historical block data. +// +// Implementations SHOULD be idempotent and safe to call multiple times for +// the same or increasing target heights. +type Pruner interface { + // PruneBlocks removes block data (header, data, signature, and hash index) + // up to and including the given height from the store, without modifying + // state snapshots or the current chain height. + PruneBlocks(ctx context.Context, height uint64) error +}