diff --git a/config/config.yaml b/config/config.yaml index 0338c2d..6279846 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -107,6 +107,7 @@ pruner: docs_per_block: 1000 # Average docs per block (~1000 on Ethereum mainnet). Pruning triggers at max_blocks * docs_per_block docs interval_seconds: 30 # How often to check and prune prune_history: true # true: each prune also deletes the removed docs' block history (walks their DAG, slower per prune); false keeps it, so the blockstore only grows. + max_docs_per_cycle: 50000 # Documents the queue drain and the height sweep each remove per cycle. Must exceed the arrival rate over one interval. schema: indexer_schema_endpoint: /api/v1/schema http_client_timeout_secs: 30 diff --git a/go.mod b/go.mod index ae4dbca..6d1e238 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/shinzonetwork/shinzo-querysig v0.2.0 github.com/shinzonetwork/viewbundle-go v0.1.1 github.com/sourcenetwork/corelog v0.0.9 - github.com/sourcenetwork/defradb v1.0.1-0.20260724174804-b05811e709e1 + github.com/sourcenetwork/defradb v1.0.1-0.20260901123234-07461cb2c4dd github.com/sourcenetwork/immutable v0.3.0 github.com/sourcenetwork/lens/host-go v0.11.0 github.com/stretchr/testify v1.12.1 diff --git a/go.sum b/go.sum index 68754c8..bf0c7de 100644 --- a/go.sum +++ b/go.sum @@ -1706,8 +1706,8 @@ github.com/sourcenetwork/corekv/namespace v0.3.1 h1:XllaUw6cteZkJYC5tgcJ40c6hIMh github.com/sourcenetwork/corekv/namespace v0.3.1/go.mod h1:uidxEQZsJ1eqecq1Zn5NipnFAgxL+VyXB6bPGziBewk= github.com/sourcenetwork/corelog v0.0.9 h1:wpoBbvju4wYtwpolfpGo8CrUVsRI8Gz1IeaXPQCW8yY= github.com/sourcenetwork/corelog v0.0.9/go.mod h1:cMabHgs3kARgYTQeQYSOmaGGP8XMU6sZrHd8LFrL3zA= -github.com/sourcenetwork/defradb v1.0.1-0.20260724174804-b05811e709e1 h1:26lOirARBiSFMISmn1KzcqBkF+LEedZyDpgpKFo0MoI= -github.com/sourcenetwork/defradb v1.0.1-0.20260724174804-b05811e709e1/go.mod h1:DItlsF6KS9fTUN7t1lnpYVjwDL3Anj8Uo9oImSUroWA= +github.com/sourcenetwork/defradb v1.0.1-0.20260901123234-07461cb2c4dd h1:YjKTsAukwIOr51Ggq5E0ejMYz8sTaEAsm6qv6kOoIAA= +github.com/sourcenetwork/defradb v1.0.1-0.20260901123234-07461cb2c4dd/go.mod h1:wrh6Vl/a4DRqVycyCn1PAYOWRKVO525H9FirHigRth0= github.com/sourcenetwork/go-libp2p-pubsub-rpc v0.0.14 h1:620zKV4rOn7U5j/WsPkk4SFj0z9/pVV4bBx0BpZQgro= github.com/sourcenetwork/go-libp2p-pubsub-rpc v0.0.14/go.mod h1:jUoQv592uUX1u7QBjAY4C+l24X9ArhPfifOqXpDHz4U= github.com/sourcenetwork/go-p2p v0.1.11 h1:ddsOsw0NTbx2b55bEP6xkNmMfh8J+FY40sQnBZrBDYI= diff --git a/pkg/host/host.go b/pkg/host/host.go index 97e4e94..43ec4df 100644 --- a/pkg/host/host.go +++ b/pkg/host/host.go @@ -489,6 +489,7 @@ func StartHostingWithEventSubscription(cfg *config.Config) (*Host, error) { //no p := pruner.NewPruner(&cfg.Pruner, defraNode) p.SetQueue(pruneQueue) + p.SetRetainHistory(cfg.HostConfig.Snapshot.Enabled) if err := p.Start(ctx); err != nil { logger.Sugar.Warnf("Failed to start pruner: %v", err) diff --git a/pkg/pruner/config.go b/pkg/pruner/config.go index 45ccd7d..060723d 100644 --- a/pkg/pruner/config.go +++ b/pkg/pruner/config.go @@ -1,5 +1,7 @@ package pruner +const defaultMaxDocsPerCycle = 50000 + // Config represents pruner configuration for removing old documents. type Config struct { Enabled bool `yaml:"enabled"` @@ -8,6 +10,9 @@ type Config struct { PruneThreshold int64 `yaml:"prune_threshold"` // Deprecated: kept for backward compatibility, unused by pruner IntervalSeconds int `yaml:"interval_seconds"` PruneHistory bool `yaml:"prune_history"` + // MaxDocsPerCycle bounds what each of the queue drain and the height sweep removes in one + // cycle. Set it above the arrival rate over one interval, or the store grows. + MaxDocsPerCycle int64 `yaml:"max_docs_per_cycle"` } // CollectionConfig defines which collections to prune and how. @@ -55,4 +60,7 @@ func (c *Config) SetDefaults() { if c.IntervalSeconds <= 0 { c.IntervalSeconds = 60 } + if c.MaxDocsPerCycle <= 0 { + c.MaxDocsPerCycle = defaultMaxDocsPerCycle + } } diff --git a/pkg/pruner/height_prune_test.go b/pkg/pruner/height_prune_test.go new file mode 100644 index 0000000..bf22578 --- /dev/null +++ b/pkg/pruner/height_prune_test.go @@ -0,0 +1,308 @@ +package pruner + +import ( + "context" + "fmt" + "testing" + + "github.com/sourcenetwork/defradb/client" + "github.com/sourcenetwork/defradb/client/options" + "github.com/sourcenetwork/defradb/node" + "github.com/stretchr/testify/require" +) + +// EventQueue.Push only accepts collection names it holds an enum for, so tests use the production +// names. +const ( + blockCollection = "Ethereum__Mainnet__Block" + logCollection = "Ethereum__Mainnet__Log" + txCollection = "Ethereum__Mainnet__Transaction" + attRecCollection = "Ethereum__Mainnet__AttestationRecord" + blockNumberColumn = "number" +) + +// heightTestSchema mirrors the shape the pruner depends on: a block collection with its own number +// field, dependents carrying blockNumber, and a dependent carrying neither. +const heightTestSchema = ` +type Ethereum__Mainnet__Block { + number: Int + hash: String +} +type Ethereum__Mainnet__Log { + blockNumber: Int + address: String +} +type Ethereum__Mainnet__Transaction { + blockNumber: Int + hash: String +} +type Ethereum__Mainnet__AttestationRecord { + attested_doc: String +} +` + +func heightTestCollections() CollectionConfig { + return CollectionConfig{ + BlockCollection: blockCollection, + BlockNumberField: blockNumberColumn, + DependentCollections: []string{logCollection, txCollection, attRecCollection}, + } +} + +// newHeightTestPruner starts a DefraDB node on a temp store and returns a pruner wired to it. +func newHeightTestPruner(t *testing.T, cfg *Config) (*Pruner, *node.Node) { + t.Helper() + ctx := context.Background() + + nb := options.Node().SetDisableAPI(true).SetDisableP2P(true) + nb.Store().SetPath(t.TempDir()) + + n, err := node.New(ctx, nb) + require.NoError(t, err) + require.NoError(t, n.Start(ctx)) + t.Cleanup(func() { _ = n.Close(ctx) }) + + _, err = n.DB.AddCollection(ctx, heightTestSchema) + require.NoError(t, err) + + cfg.SetDefaults() + p := NewPruner(cfg, n, heightTestCollections()) + p.heightPrunable = p.resolveHeightPrunable(ctx) + return p, n +} + +func addHeightDoc(t *testing.T, n *node.Node, collection string, fields map[string]any) { + t.Helper() + ctx := context.Background() + col, err := n.DB.GetCollectionByName(ctx, collection) + require.NoError(t, err) + doc, err := client.NewDocFromMap(ctx, fields, col.Version()) + require.NoError(t, err) + require.NoError(t, col.AddDocument(ctx, doc)) +} + +// blockNumbers returns fieldName across a collection, so a test can assert which documents +// survived rather than only how many. +func blockNumbers(t *testing.T, n *node.Node, collection, fieldName string) []int64 { + t.Helper() + res := n.DB.ExecRequest(context.Background(), + fmt.Sprintf("query { %s(order: {%s: ASC}) { %s } }", collection, fieldName, fieldName)) + require.Empty(t, res.GQL.Errors) + + data, ok := res.GQL.Data.(map[string]any) + require.True(t, ok) + + var out []int64 + switch docs := data[collection].(type) { + case []map[string]any: + for _, d := range docs { + n, err := parseBlockNumber(d[fieldName]) + require.NoError(t, err) + out = append(out, n) + } + case []any: + for _, raw := range docs { + d, ok := raw.(map[string]any) + require.True(t, ok) + n, err := parseBlockNumber(d[fieldName]) + require.NoError(t, err) + out = append(out, n) + } + } + return out +} + +func countHeightDocs(t *testing.T, n *node.Node, collection string) int { + t.Helper() + res := n.DB.ExecRequest(context.Background(), fmt.Sprintf("query { %s { _docID } }", collection)) + require.Empty(t, res.GQL.Errors) + data, ok := res.GQL.Data.(map[string]any) + require.True(t, ok) + switch docs := data[collection].(type) { + case []map[string]any: + return len(docs) + case []any: + return len(docs) + } + return 0 +} + +// seedHeightBlocks writes one Block and one Log per block number in [from, to]. +func seedHeightBlocks(t *testing.T, n *node.Node, from, to int) { + t.Helper() + for i := from; i <= to; i++ { + addHeightDoc(t, n, blockCollection, map[string]any{"number": i, "hash": fmt.Sprintf("h%d", i)}) + addHeightDoc(t, n, logCollection, map[string]any{"blockNumber": i, "address": fmt.Sprintf("a%d", i)}) + } +} + +// A restart leaves the store holding documents the queue never recorded. +func TestPruneRemovesDocumentsTheQueueNeverSaw(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000}) + p.SetQueue(NewEventQueue(heightTestCollections())) + + seedHeightBlocks(t, n, 1, 20) + + require.NoError(t, p.runPrune(context.Background())) + + require.Equal(t, []int64{16, 17, 18, 19, 20}, blockNumbers(t, n, blockCollection, blockNumberColumn)) + require.Equal(t, []int64{16, 17, 18, 19, 20}, blockNumbers(t, n, logCollection, dependentBlockNumberField)) +} + +// A dependent collection can hold blocks the block collection has already dropped. +func TestPruneRemovesDependentTailBelowTheWindow(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000}) + p.SetQueue(NewEventQueue(heightTestCollections())) + + for i := 16; i <= 20; i++ { + addHeightDoc(t, n, blockCollection, map[string]any{"number": i, "hash": fmt.Sprintf("h%d", i)}) + } + for i := 1; i <= 20; i++ { + addHeightDoc(t, n, logCollection, map[string]any{"blockNumber": i, "address": fmt.Sprintf("a%d", i)}) + } + + require.NoError(t, p.runPrune(context.Background())) + + require.Equal(t, []int64{16, 17, 18, 19, 20}, blockNumbers(t, n, blockCollection, blockNumberColumn)) + require.Equal(t, []int64{16, 17, 18, 19, 20}, blockNumbers(t, n, logCollection, dependentBlockNumberField)) +} + +// Block zero is a real block number, not an empty collection. +func TestPruneHandlesBlockZero(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000}) + p.SetQueue(NewEventQueue(heightTestCollections())) + + seedHeightBlocks(t, n, 0, 20) + + require.NoError(t, p.runPrune(context.Background())) + + require.Equal(t, []int64{16, 17, 18, 19, 20}, blockNumbers(t, n, blockCollection, blockNumberColumn)) + require.Equal(t, []int64{16, 17, 18, 19, 20}, blockNumbers(t, n, logCollection, dependentBlockNumberField)) +} + +// A queue far enough over its threshold to spend the whole drain budget must still leave the +// sweep able to run. +func TestHeightSweepRunsWhenTheDrainSpendsItsBudget(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{ + Enabled: true, MaxBlocks: 5, DocsPerBlock: 1, MaxDocsPerCycle: 4, + }) + q := NewEventQueue(heightTestCollections()) + p.SetQueue(q) + + seedHeightBlocks(t, n, 1, 20) + for i := range 9 { + q.Push(logCollection, testDocID(i)) + } + + require.NoError(t, p.runPrune(context.Background())) + + // 9 queued against a threshold of 5, capped at 4. + require.Equal(t, 5, q.Len()) + // The cutoff is 15, and the sweep spends its own 4 on the oldest logs. + require.Equal(t, []int64{5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}, + blockNumbers(t, n, logCollection, dependentBlockNumberField)) + // Blocks come last and the sweep budget is gone by then. + require.Len(t, blockNumbers(t, n, blockCollection, blockNumberColumn), 20) +} + +// The sweep stops once the cycle's budget is spent, however far below the window the store is. +func TestHeightSweepStopsAtTheCycleBudget(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{ + Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000, MaxDocsPerCycle: 3, + }) + p.SetQueue(NewEventQueue(heightTestCollections())) + + seedHeightBlocks(t, n, 1, 20) + + require.NoError(t, p.runPrune(context.Background())) + + require.Len(t, blockNumbers(t, n, logCollection, dependentBlockNumberField), 17) + require.Len(t, blockNumbers(t, n, blockCollection, blockNumberColumn), 20) +} + +// The budget is spent across collections in order: a collection that needs less than the remainder +// leaves the rest for the next one. +func TestHeightSweepBudgetIsSharedAcrossCollections(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{ + Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000, MaxDocsPerCycle: 5, + }) + p.SetQueue(NewEventQueue(heightTestCollections())) + + for i := 1; i <= 20; i++ { + addHeightDoc(t, n, blockCollection, map[string]any{"number": i, "hash": fmt.Sprintf("h%d", i)}) + addHeightDoc(t, n, txCollection, map[string]any{"blockNumber": i, "hash": fmt.Sprintf("t%d", i)}) + } + // Only two Log rows sit below the cutoff of 15, so Log cannot use the whole budget. + for _, i := range []int{14, 15} { + addHeightDoc(t, n, logCollection, map[string]any{"blockNumber": i, "address": fmt.Sprintf("a%d", i)}) + } + + require.NoError(t, p.runPrune(context.Background())) + + require.Empty(t, blockNumbers(t, n, logCollection, dependentBlockNumberField)) + require.Equal(t, []int64{4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}, + blockNumbers(t, n, txCollection, dependentBlockNumberField)) + require.Len(t, blockNumbers(t, n, blockCollection, blockNumberColumn), 20) +} + +// Zero is unlimited to the query planner, so a spent budget must remove nothing rather than +// everything. +func TestPurgeCollectionBelowRemovesNothingWithoutBudget(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000}) + seedHeightBlocks(t, n, 1, 20) + + purged, err := p.purgeCollectionBelow(context.Background(), logCollection, dependentBlockNumberField, 15, 0) + require.NoError(t, err) + require.Zero(t, purged) + require.Len(t, blockNumbers(t, n, logCollection, dependentBlockNumberField), 20) +} + +// A collection with no block-number field cannot be ordered by height, so it is left alone. +func TestHeightPruneSkipsCollectionWithoutBlockNumber(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000}) + p.SetQueue(NewEventQueue(heightTestCollections())) + + require.Equal(t, []string{logCollection, txCollection}, p.heightPrunable) + + seedHeightBlocks(t, n, 1, 20) + for i := 1; i <= 3; i++ { + addHeightDoc(t, n, attRecCollection, map[string]any{"attested_doc": fmt.Sprintf("d%d", i)}) + } + + require.NoError(t, p.runPrune(context.Background())) + + require.Equal(t, 3, countHeightDocs(t, n, attRecCollection)) +} + +// A node bootstrapped with historical blocks keeps them. +func TestRetainHistorySuppressesHeightPrune(t *testing.T) { + p, n := newHeightTestPruner(t, &Config{Enabled: true, MaxBlocks: 5, DocsPerBlock: 1000}) + p.SetQueue(NewEventQueue(heightTestCollections())) + p.SetRetainHistory(true) + + seedHeightBlocks(t, n, 1, 20) + + require.NoError(t, p.runPrune(context.Background())) + + require.Len(t, blockNumbers(t, n, blockCollection, blockNumberColumn), 20) + require.Len(t, blockNumbers(t, n, logCollection, dependentBlockNumberField), 20) +} + +// One cycle removes at most MaxDocsPerCycle, however far behind the queue is. +func TestDrainQueueStopsAtThePerCycleLimit(t *testing.T) { + p, _ := newHeightTestPruner(t, &Config{ + Enabled: true, MaxBlocks: 1, DocsPerBlock: 10, MaxDocsPerCycle: 25, + }) + q := NewEventQueue(heightTestCollections()) + p.SetQueue(q) + + for i := range 200 { + q.Push(logCollection, testDocID(i)) + } + + require.NoError(t, p.drainQueue(context.Background(), q)) + require.Equal(t, 175, q.Len()) + + require.NoError(t, p.drainQueue(context.Background(), q)) + require.Equal(t, 150, q.Len()) +} diff --git a/pkg/pruner/pruner.go b/pkg/pruner/pruner.go index ae2ea76..f97b1a9 100644 --- a/pkg/pruner/pruner.go +++ b/pkg/pruner/pruner.go @@ -22,6 +22,10 @@ const ( purgeProgressInterval = 30 * time.Second ) +// dependentBlockNumberField names the block a dependent document belongs to. The block collection +// names its own field through CollectionConfig. +const dependentBlockNumberField = "blockNumber" + // errStopped ends a purge early because the pruner is shutting down. Its documents are // re-queued, so the work resumes rather than being lost. var errStopped = errors.New("pruner stopped") @@ -47,9 +51,13 @@ type Pruner struct { collections CollectionConfig defraNode *node.Node queue PrunerQueue // EventQueue (the only implementation in host) - stopChan chan struct{} - wg sync.WaitGroup - mu sync.RWMutex + // retainHistory disables the height sweep, for a node bootstrapped with history it should keep. + retainHistory bool + // heightPrunable is the subset of DependentCollections carrying dependentBlockNumberField. + heightPrunable []string + stopChan chan struct{} + wg sync.WaitGroup + mu sync.RWMutex // purgeDocs deletes one batch of documents. Set only by tests; in production it is nil // and the collection's own PurgeByDocIDs is used, since a purge is otherwise only @@ -91,6 +99,11 @@ func (p *Pruner) SetQueue(queue PrunerQueue) { p.queue = queue } +// SetRetainHistory keeps blocks below the retention window instead of pruning them by height. +func (p *Pruner) SetRetainHistory(retain bool) { + p.retainHistory = retain +} + // Start begins the pruning loop in a background goroutine. func (p *Pruner) Start(ctx context.Context) error { if !p.cfg.Enabled { @@ -181,6 +194,8 @@ func (p *Pruner) GetMetrics() Metrics { func (p *Pruner) pruneLoop(ctx context.Context) { defer p.wg.Done() + p.heightPrunable = p.resolveHeightPrunable(ctx) + // Run startup cleanup only for indexer queues (no P2P) or when no queue is set. // For event queues (hosts), skip startup cleanup — the DB may contain snapshot- // imported data that should not be pruned. Only queue-tracked data gets pruned. @@ -211,17 +226,43 @@ func (p *Pruner) pruneLoop(ctx context.Context) { } } +// resolveHeightPrunable returns the dependent collections the height sweep can order on. One +// without the field is bounded only by the queue. +func (p *Pruner) resolveHeightPrunable(ctx context.Context) []string { + prunable := make([]string, 0, len(p.collections.DependentCollections)) + var skipped []string + + for _, name := range p.collections.DependentCollections { + col, err := p.defraNode.DB.GetCollectionByName(ctx, name) + if err != nil { + skipped = append(skipped, name) + continue + } + if _, ok := col.Version().GetFieldByName(dependentBlockNumberField); !ok { + skipped = append(skipped, name) + continue + } + prunable = append(prunable, name) + } + + if len(skipped) > 0 { + logger.Sugar.Warnf("Height prune skips %v: no %s field, so these are bounded only by the queue", + skipped, dependentBlockNumberField) + } + return prunable +} + // runPrune executes the appropriate pruning strategy based on queue type and state. func (p *Pruner) runPrune(ctx context.Context) error { if p.queue == nil { - return p.filterBasedPrune(ctx) + return p.pruneBeyondRetention(ctx, p.cfg.MaxDocsPerCycle) } switch q := p.queue.(type) { case *EventQueue: return p.runEventQueuePrune(ctx, q) default: - return p.filterBasedPrune(ctx) + return p.pruneBeyondRetention(ctx, p.cfg.MaxDocsPerCycle) } } @@ -230,14 +271,23 @@ func (p *Pruner) runPrune(ctx context.Context) error { // arrive in non-deterministic order — block docs may arrive before their // dependent docs (transactions, logs, etc.). func (p *Pruner) runEventQueuePrune(ctx context.Context, q *EventQueue) error { + if err := p.drainQueue(ctx, q); err != nil { + return err + } + if p.retainHistory { + return nil + } + // Budgeted separately from the drain: the two select different documents, and a queue far + // enough over its threshold would otherwise leave the sweep nothing. + return p.pruneBeyondRetention(ctx, p.cfg.MaxDocsPerCycle) +} + +// drainQueue removes the queue's excess over max_docs, within the cycle's budget. +func (p *Pruner) drainQueue(ctx context.Context, q *EventQueue) error { totalDocs := int64(q.Len()) maxDocs := p.cfg.MaxDocs() if totalDocs <= maxDocs { - // Queue is underfilled (e.g., after a crash restart where the queue was lost). - // Do NOT fall back to filter-based pruning — the DB may contain snapshot- - // imported data that should not be pruned. Only prune what the queue tracks. - // // Logged so a queue that never reaches the threshold is distinguishable from a // pruner that is not running. logger.Sugar.Infof("Prune skipped: queue has %d docs, threshold %d (max_blocks=%d × docs_per_block=%d)", @@ -245,8 +295,8 @@ func (p *Pruner) runEventQueuePrune(ctx context.Context, q *EventQueue) error { return nil } - excess := int(totalDocs - maxDocs) - result := q.DrainDocs(excess) + excess := min(totalDocs-maxDocs, p.cfg.MaxDocsPerCycle) + result := q.DrainDocs(int(excess)) if result == nil { return nil } @@ -348,7 +398,7 @@ func (p *Pruner) startupCleanup(ctx context.Context) error { logger.Sugar.Infof("Startup cleanup: pruning blocks %d-%d (%d blocks, keeping %d-%d)", lowest, cutoffBlock, toPrune, cutoffBlock+1, highest) - totalSubmitted, err := p.pruneBlockRange(ctx, lowest, cutoffBlock) + totalSubmitted, blocksPruned, err := p.pruneBelow(ctx, cutoffBlock, p.cfg.MaxDocsPerCycle) if err != nil { logger.Sugar.Errorf("Startup: failed to prune blocks %d-%d: %v", lowest, cutoffBlock, err) return err @@ -357,7 +407,7 @@ func (p *Pruner) startupCleanup(ctx context.Context) error { logger.Sugar.Infof("Startup cleanup complete: submitted %d documents", totalSubmitted) p.mu.Lock() - p.totalBlocksPruned += toPrune + p.totalBlocksPruned += blocksPruned p.totalDocsSubmitted += totalSubmitted p.lastPruneTime = time.Now() p.mu.Unlock() @@ -365,43 +415,35 @@ func (p *Pruner) startupCleanup(ctx context.Context) error { return nil } -// filterBasedPrune checks the actual DB block count and prunes excess blocks. -// Used by the indexer queue (no P2P) and as a fallback when the queue is underfilled. -func (p *Pruner) filterBasedPrune(ctx context.Context) error { - highest, err := p.getHighestBlockNumber(ctx) - if err != nil { - return err - } - if highest == 0 { +// pruneBeyondRetention removes documents for blocks below the retention window, whether or not the +// queue knows about them, within the budget left for this cycle. The cutoff is measured from the +// highest block the node holds. +func (p *Pruner) pruneBeyondRetention(ctx context.Context, budget int64) error { + if budget <= 0 { return nil } - lowest, err := p.getLowestBlockNumber(ctx) + highest, err := p.getHighestBlockNumber(ctx) if err != nil { return err } - if lowest == 0 { - return nil - } - dbBlockCount := highest - lowest + 1 - if dbBlockCount <= p.cfg.MaxBlocks { + cutoff := highest - p.cfg.MaxBlocks + if cutoff <= 0 { + // The node holds no more than the retention window, including when the store is empty. return nil } - excess := dbBlockCount - p.cfg.MaxBlocks - cutoff := lowest + excess - 1 - - logger.Sugar.Infof("Filter-based prune: %d excess blocks (%d-%d), pruning %d-%d", - excess, lowest, highest, lowest, cutoff) - - submitted, err := p.pruneBlockRange(ctx, lowest, cutoff) + submitted, blocks, err := p.pruneBelow(ctx, cutoff, budget) if err != nil { return err } + if submitted == 0 { + return nil + } p.mu.Lock() - p.totalBlocksPruned += excess + p.totalBlocksPruned += blocks p.totalDocsSubmitted += submitted p.lastPruneTime = time.Now() p.mu.Unlock() @@ -409,54 +451,78 @@ func (p *Pruner) filterBasedPrune(ctx context.Context) error { return nil } -// pruneBlockRange removes all documents for blocks in [startBlock, endBlock]. -// Uses order+limit queries to get docIDs, then purges them. -// Safe to call with concurrent P2P replication — merge handles missing blocks gracefully. -func (p *Pruner) pruneBlockRange(ctx context.Context, startBlock, endBlock int64) (int64, error) { - totalSubmitted := int64(0) - - logger.Sugar.Infof("pruneBlockRange: deleting blocks %d-%d (%d blocks)", - startBlock, endBlock, endBlock-startBlock+1) - - // Dependent collections first, block collection last - for _, colName := range p.collections.DependentCollections { - docIDs, err := p.queryOldestDocIDs(ctx, colName, "blockNumber", endBlock) +// pruneBelow removes documents at or below cutoff, dependent collections before the block +// collection, so a block is not removed ahead of the documents that reference it. A stop ends the +// cycle where it is; what is left is found again by the next one. +// +// Safe to run alongside P2P replication: a merge for a removed block is handled as a new document. +func (p *Pruner) pruneBelow(ctx context.Context, cutoff, budget int64) (submitted, blocks int64, err error) { + for _, colName := range p.heightPrunable { + purged, err := p.purgeCollectionBelow(ctx, colName, dependentBlockNumberField, cutoff, budget-submitted) if err != nil { - logger.Sugar.Warnf("pruneBlockRange: query failed for %s (skipping): %v", colName, err) + if abandoned(err) { + return submitted, blocks, nil + } + logger.Sugar.Warnf("Prune below %d: %s skipped: %v", cutoff, colName, err) continue } - if len(docIDs) > 0 { - submitted, err := p.purgeByDocIDs(ctx, colName, docIDs) - if err != nil { - logger.Sugar.Warnf("pruneBlockRange: failed to purge %s: %v", colName, err) - } else { - totalSubmitted += submitted - } + submitted += purged + } + + blocks, err = p.purgeCollectionBelow(ctx, p.collections.BlockCollection, p.collections.BlockNumberField, cutoff, budget-submitted) + if err != nil { + if abandoned(err) { + return submitted, 0, nil } + return submitted, 0, fmt.Errorf("prune below %d: %s: %w", cutoff, p.collections.BlockCollection, err) + } + submitted += blocks + + if submitted > 0 { + logger.Sugar.Infof("Prune below %d: submitted %d documents across %d blocks", cutoff, submitted, blocks) } + return submitted, blocks, nil +} + +// abandoned reports whether an error ended the work rather than failed it, so the caller stops +// instead of moving on to the next collection. +func abandoned(err error) bool { + return errors.Is(err, errStopped) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) +} - blockDocIDs, err := p.queryOldestDocIDs(ctx, p.collections.BlockCollection, p.collections.BlockNumberField, endBlock) +// purgeCollectionBelow removes one collection's documents at or below cutoff, up to the query +// limit. Each collection is checked on its own, because a dependent can hold older blocks than the +// block collection does. +func (p *Pruner) purgeCollectionBelow(ctx context.Context, collectionName, fieldName string, cutoff, limit int64) (int64, error) { + oldest, found, err := p.edgeBlockNumber(ctx, collectionName, fieldName, "ASC") if err != nil { - return totalSubmitted, fmt.Errorf("query failed for blocks: %w", err) + return 0, err } - if len(blockDocIDs) > 0 { - submitted, err := p.purgeByDocIDs(ctx, p.collections.BlockCollection, blockDocIDs) - if err != nil { - return totalSubmitted, fmt.Errorf("failed to purge blocks: %w", err) - } - totalSubmitted += submitted + if !found || oldest > cutoff { + return 0, nil + } + + docIDs, err := p.queryOldestDocIDs(ctx, collectionName, fieldName, cutoff, limit) + if err != nil { + return 0, err + } + if len(docIDs) == 0 { + return 0, nil } - logger.Sugar.Infof("pruneBlockRange: submitted %d docs for blocks %d-%d", totalSubmitted, startBlock, endBlock) - return totalSubmitted, nil + return p.purgeByDocIDs(ctx, collectionName, docIDs) } // ─── Document operations ───────────────────────────────────────────────────── // queryOldestDocIDs queries for docIDs where fieldName <= maxBlockNumber using order+limit. // Works on P2P-replicated data where filter queries return empty results. -func (p *Pruner) queryOldestDocIDs(ctx context.Context, collectionName, fieldName string, maxBlockNumber int64) ([]string, error) { - limit := 50000 +func (p *Pruner) queryOldestDocIDs(ctx context.Context, collectionName, fieldName string, maxBlockNumber, limit int64) ([]string, error) { + // A limit of zero is unlimited to the query planner, so a spent budget stops here. + if limit <= 0 { + return nil, nil + } + query := fmt.Sprintf(`query { %s(order: { %s: ASC }, limit: %d) { _docID @@ -584,67 +650,59 @@ func (p *Pruner) purgeByDocIDs(ctx context.Context, collectionName string, docID // ─── Block number queries ──────────────────────────────────────────────────── func (p *Pruner) getLowestBlockNumber(ctx context.Context) (int64, error) { - query := `query { - ` + p.collections.BlockCollection + ` (order: {` + p.collections.BlockNumberField + `: ASC}, limit: 1) { - ` + p.collections.BlockNumberField + ` - } - }` - - result := p.defraNode.DB.ExecRequest(ctx, query) - if len(result.GQL.Errors) > 0 { - return 0, result.GQL.Errors[0] - } - - return p.extractBlockNumber(result.GQL.Data) + lowest, _, err := p.edgeBlockNumber(ctx, p.collections.BlockCollection, p.collections.BlockNumberField, "ASC") + return lowest, err } func (p *Pruner) getHighestBlockNumber(ctx context.Context) (int64, error) { - query := `query { - ` + p.collections.BlockCollection + ` (order: {` + p.collections.BlockNumberField + `: DESC}, limit: 1) { - ` + p.collections.BlockNumberField + ` + highest, _, err := p.edgeBlockNumber(ctx, p.collections.BlockCollection, p.collections.BlockNumberField, "DESC") + return highest, err +} + +// edgeBlockNumber reads the block number at one end of a collection's ordering. The bool is false +// when the collection is empty, which a zero block number cannot be distinguished from otherwise. +func (p *Pruner) edgeBlockNumber(ctx context.Context, collectionName, fieldName, direction string) (int64, bool, error) { + query := fmt.Sprintf(`query { + %s(order: { %s: %s }, limit: 1) { + %s } - }` + }`, collectionName, fieldName, direction, fieldName) result := p.defraNode.DB.ExecRequest(ctx, query) if len(result.GQL.Errors) > 0 { - return 0, result.GQL.Errors[0] + return 0, false, result.GQL.Errors[0] } - return p.extractBlockNumber(result.GQL.Data) + return extractBlockNumber(result.GQL.Data, collectionName, fieldName) } -func (p *Pruner) extractBlockNumber(gqlData any) (int64, error) { +func extractBlockNumber(gqlData any, collectionName, fieldName string) (int64, bool, error) { data, ok := gqlData.(map[string]any) if !ok { - return 0, nil + return 0, false, nil } - blocksRaw := data[p.collections.BlockCollection] - - if blocksTyped, ok := blocksRaw.([]map[string]any); ok { - if len(blocksTyped) == 0 { - return 0, nil + // DefraDB returns []map[string]any or []any depending on context; both reach here. + var first map[string]any + switch docs := data[collectionName].(type) { + case []map[string]any: + if len(docs) == 0 { + return 0, false, nil } - if number, ok := blocksTyped[0][p.collections.BlockNumberField]; ok { - return parseBlockNumber(number) + first = docs[0] + case []any: + if len(docs) == 0 { + return 0, false, nil } - return 0, nil - } - - blocks, ok := blocksRaw.([]any) - if !ok || len(blocks) == 0 { - return 0, nil - } - - block, ok := blocks[0].(map[string]any) - if !ok { - return 0, nil + if first, ok = docs[0].(map[string]any); !ok { + return 0, false, nil + } + default: + return 0, false, nil } - if number, ok := block[p.collections.BlockNumberField]; ok { - return parseBlockNumber(number) - } - return 0, nil + number, err := parseBlockNumber(first[fieldName]) + return number, err == nil, err } func parseBlockNumber(number any) (int64, error) {