fix: prune by block height so a lost queue does not stop pruning

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

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

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=

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)

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 @@
 	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.
@@ -27,7 +32,7 @@
 func DefaultCollectionConfig() CollectionConfig {
 	return CollectionConfig{
 		BlockCollection:  "Ethereum__Mainnet__Block",
 		BlockNumberField: "number",
 		DependentCollections: []string{
 			"Ethereum__Mainnet__BatchSignature",
 			"Ethereum__Mainnet__AccessListEntry",
@@ -55,4 +60,7 @@
 	if c.IntervalSeconds <= 0 {
 		c.IntervalSeconds = 60
 	}
+	if c.MaxDocsPerCycle <= 0 {
+		c.MaxDocsPerCycle = defaultMaxDocsPerCycle
+	}
 }

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())
+}