chanevents: add channel event pruning

We add regular channel event pruning, as otherwise the database may get
filled quickly. We add two mechanisms, a retention time and a max events
number. Both can be turned on individually.

chanevents/chanevents.go

@@ -9,6 +9,22 @@ import (
 	"github.com/lightningnetwork/lnd/fn/v2"
 )
 
+// Config holds the configuration options for channel event pruning. See the
+// README for storage sizing guidance.
+type Config struct {
+	// MaxEvents is the maximum number of channel events to retain. Once the
+	// table exceeds this count, the oldest events are pruned. This operates
+	// as a hard ceiling on database size to prevent disk filling. A value
+	// of 0 disables this limit.
+	MaxEvents uint64 `long:"max-events" description:"The maximum number of channel events to retain before pruning the oldest events. This limit acts as a hard ceiling to prevent disk filling. A value of 0 disables pruning based on the number of events."`
+
+	// Retention is the minimum duration of channel events to keep. Events
+	// older than this window are pruned, even if the max-events limit is
+	// not exceeded. If max-events is exceeded, newer events can still be
+	// pruned to enforce the size ceiling. A value of 0 disables this limit.
+	Retention time.Duration `long:"retention" description:"The minimum duration of channel events to keep. Events older than this window are pruned, even if the max-events limit is not exceeded. A value of 0 disables pruning based on age."`
+}
+
 // EventType is an enum for the different types of channel events.
 type EventType int16
 

chanevents/monitor.go

@@ -18,6 +18,16 @@ const (
 	// retryInterval is the time to wait before retrying after a
 	// transient error or while waiting for lnd to become ready.
 	retryInterval = 5 * time.Second
+
+	// pruneInterval is how often the monitor enforces the channel event
+	// storage limits while consuming live events.
+	pruneInterval = time.Hour
+
+	// minPruneInterval is the floor for the background pruning ticker. A
+	// tiny retention window would otherwise drive the ticker interval down
+	// to milliseconds and starve the CPU, so we never tick faster than
+	// this.
+	minPruneInterval = time.Second
 )
 
 var (
@@ -42,15 +52,25 @@ type Monitor struct {
 	// channel events.
 	store *Store
 
+	// cfg holds the channel event pruning configuration.
+	cfg Config
+
+	// warnedDestructivePrune ensures the operator is warned only once that
+	// pruning has permanently deleted events.
+	warnedDestructivePrune atomic.Bool
+
 	wg   sync.WaitGroup
 	quit chan struct{}
 }
 
 // NewMonitor creates a new channel events monitor.
-func NewMonitor(lnd lndclient.LightningClient, store *Store) *Monitor {
+func NewMonitor(lnd lndclient.LightningClient, store *Store,
+	cfg Config) *Monitor {
+
 	return &Monitor{
 		lnd:   lnd,
 		store: store,
+		cfg:   cfg,
 		quit:  make(chan struct{}),
 	}
 }
@@ -94,6 +114,28 @@ func (m *Monitor) monitorLoop(ctx context.Context) {
 
 	log.Info("Channel events monitor starting")
 
+	// Prune periodically while consuming live events, to bound the query
+	// overhead on high-frequency channels. Only arm the ticker when pruning
+	// is actually enabled; otherwise leave pruneChan nil so the select below
+	// never fires and we don't spin a ticker for nothing.
+	var pruneChan <-chan time.Time
+	if m.cfg.MaxEvents > 0 || m.cfg.Retention > 0 {
+		pruneIntervalToUse := pruneInterval
+		if m.cfg.Retention > 0 && m.cfg.Retention < pruneIntervalToUse {
+			pruneIntervalToUse = m.cfg.Retention
+
+			// Never tick faster than the floor: an extremely small
+			// retention would otherwise spin the ticker continuously.
+			if pruneIntervalToUse < minPruneInterval {
+				pruneIntervalToUse = minPruneInterval
+			}
+		}
+
+		pruneTicker := time.NewTicker(pruneIntervalToUse)
+		defer pruneTicker.Stop()
+		pruneChan = pruneTicker.C
+	}
+
 	var synced bool
 
 	for {
@@ -108,12 +150,16 @@ func (m *Monitor) monitorLoop(ctx context.Context) {
 				log.Errorf("Error during initial sync: %v", err)
 			} else {
 				synced = true
+
+				// The initial sync can insert a sizeable number
+				// of events, so prune once it completes.
+				m.pruneEvents(ctx)
 			}
 		}
 
 		// Subscribe and consume events until the stream breaks or an
 		// error occurs.
-		if !m.subscribe(ctx) {
+		if !m.subscribe(ctx, pruneChan) {
 			return
 		}
 
@@ -130,6 +176,36 @@ func (m *Monitor) monitorLoop(ctx context.Context) {
 	}
 }
 
+// pruneEvents enforces the configured channel event storage limits, logging
+// how many events were deleted.
+func (m *Monitor) pruneEvents(ctx context.Context) {
+	pruned, err := m.store.PruneEvents(
+		ctx, m.cfg.MaxEvents, m.cfg.Retention,
+	)
+	if err != nil {
+		log.Errorf("Error pruning channel events: %v", err)
+		return
+	}
+
+	if pruned == 0 {
+		return
+	}
+
+	// Pruning permanently deletes events, so warn the first time it happens.
+	// This gives a clear signal to an operator who did not expect the default
+	// limits to remove pre-existing history. Subsequent prunes log at info
+	// level.
+	if m.warnedDestructivePrune.CompareAndSwap(false, true) {
+		log.Warnf("Pruned %d channel event(s) to enforce storage "+
+			"limits (max-events=%d, retention=%v); pruning is "+
+			"enabled by default and permanently deletes events",
+			pruned, m.cfg.MaxEvents, m.cfg.Retention)
+	} else {
+		log.Infof("Pruned %d channel event(s) to enforce storage "+
+			"limits", pruned)
+	}
+}
+
 // waitForReady polls lnd's GetInfo until it reports SyncedToChain. It retries
 // on transient RPC errors. It returns true when lnd is ready, or false if the
 // monitor is shutting down.
@@ -157,7 +233,9 @@ func (m *Monitor) waitForReady(ctx context.Context) bool {
 // subscribe subscribes to lnd channel events and processes them until the
 // stream breaks or an error occurs. It returns true on transient failures
 // (caller should retry) or false if the monitor is shutting down.
-func (m *Monitor) subscribe(ctx context.Context) bool {
+func (m *Monitor) subscribe(ctx context.Context,
+	pruneChan <-chan time.Time) bool {
+
 	eventChan, errChan, err := m.lnd.SubscribeChannelEvents(ctx)
 	if err != nil {
 		log.Errorf("Error subscribing to channel events: %v", err)
@@ -178,6 +256,11 @@ func (m *Monitor) subscribe(ctx context.Context) bool {
 					err)
 			}
 
+		case <-pruneChan:
+			// Periodically enforce the storage limits to keep the
+			// channel_events table bounded.
+			m.pruneEvents(ctx)
+
 		case err, ok := <-errChan:
 			if !ok {
 				log.Warn("Channel event error stream " +

chanevents/store.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/btcsuite/btcd/btcutil"
@@ -51,6 +52,12 @@ type Queries interface {
 	)
 
 	GetChannels(ctx context.Context) ([]sqlc.GetChannelsRow, error)
+
+	PruneChannelEventsBySize(ctx context.Context, offset int32) (int64,
+		error)
+
+	PruneChannelEventsByAge(ctx context.Context, timestamp time.Time) (
+		int64, error)
 }
 
 // Store provides access to the db for channel events.
@@ -337,6 +344,60 @@ func (s *Store) GetLatestChannelUpdateBefore(ctx context.Context,
 	return marshalChannelEvent(dbEvent), nil
 }
 
+// PruneEvents enforces the size and age storage limits independently,
+// returning the number of events deleted. A zero maxEvents or retention
+// disables the corresponding limit, and zero for both disables pruning.
+func (s *Store) PruneEvents(ctx context.Context, maxEvents uint64,
+	retention time.Duration) (int64, error) {
+
+	// If both options are 0, pruning is completely disabled.
+	if maxEvents == 0 && retention == 0 {
+		return 0, nil
+	}
+
+	var pruned int64
+
+	// Enforce the size ceiling by keeping only the newest maxEvents rows.
+	// An offset of (maxEvents - 1) lands on the oldest row we want to keep,
+	// so everything with a smaller id is deleted.
+	if maxEvents > 0 {
+		// The size limit becomes an int32 SQL OFFSET below. ValidateConfig
+		// already rejects an out-of-range max-events, but it is not run on
+		// every initialization path (e.g. when faraday runs as a
+		// subserver), so guard the cast here too: an overflowing value
+		// would wrap to a tiny offset and prune almost the entire table.
+		if maxEvents > math.MaxInt32 {
+			return pruned, fmt.Errorf("maxEvents %d exceeds maximum "+
+				"allowed value %d", maxEvents, math.MaxInt32)
+		}
+
+		bySize, err := s.db.PruneChannelEventsBySize(
+			ctx, int32(maxEvents-1),
+		)
+		if err != nil {
+			return pruned, fmt.Errorf("failed to prune channel "+
+				"events by size: %w", err)
+		}
+
+		pruned += bySize
+	}
+
+	// Enforce the retention window by deleting anything older than the
+	// cutoff.
+	if retention > 0 {
+		cutoff := s.clock.Now().UTC().Add(-retention)
+		byAge, err := s.db.PruneChannelEventsByAge(ctx, cutoff)
+		if err != nil {
+			return pruned, fmt.Errorf("failed to prune channel "+
+				"events by age: %w", err)
+		}
+
+		pruned += byAge
+	}
+
+	return pruned, nil
+}
+
 // marshalChannelEvent converts a db channel event into our internal type.
 func marshalChannelEvent(dbEvent sqlc.ChannelEvent) *ChannelEvent {
 	var localBalance fn.Option[btcutil.Amount]

chanevents/store_test.go

@@ -196,6 +196,237 @@ func TestStore(t *testing.T) {
 	)
 }
 
+// pruneFixture is an isolated environment for a single TestPruneEvents case. It
+// holds a fresh store with two channels and a fixed clock, and exposes helpers
+// to seed events and inspect the table without leaking state between cases.
+type pruneFixture struct {
+	t      *testing.T
+	store  *Store
+	ctx    context.Context
+	chan1  int64
+	chan2  int64
+	now    time.Time
+	old    time.Time
+	recent time.Time
+}
+
+// newPruneFixture builds a fresh store with two channels on one peer and pins
+// the clock to a reference point. It derives an "old" timestamp well outside
+// and a "recent" timestamp well inside a 30-day retention window.
+func newPruneFixture(t *testing.T) *pruneFixture {
+	t.Helper()
+
+	clk := clock.NewTestClock(testTime)
+	store := NewTestDB(t, clk)
+	ctx := context.Background()
+
+	peerID, err := store.AddPeer(ctx, testPubKey)
+	require.NoError(t, err)
+
+	chan1, err := store.AddChannel(
+		ctx, testChanPoint1, testShortChanID1, peerID,
+	)
+	require.NoError(t, err)
+
+	chan2, err := store.AddChannel(
+		ctx, testChanPoint2, testShortChanID2, peerID,
+	)
+	require.NoError(t, err)
+
+	now := testTime.Add(100 * 24 * time.Hour)
+	clk.SetTime(now)
+
+	return &pruneFixture{
+		t:      t,
+		store:  store,
+		ctx:    ctx,
+		chan1:  chan1,
+		chan2:  chan2,
+		now:    now,
+		old:    now.Add(-90 * 24 * time.Hour),
+		recent: now.Add(-5 * 24 * time.Hour),
+	}
+}
+
+// addEvents inserts n update events on the given channel at timestamp ts.
+func (f *pruneFixture) addEvents(channelID int64, ts time.Time, n int) {
+	f.t.Helper()
+
+	for i := 0; i < n; i++ {
+		err := f.store.AddChannelEvent(f.ctx, &ChannelEvent{
+			ChannelID: channelID,
+			EventType: EventTypeUpdate,
+			Timestamp: ts,
+		})
+		require.NoError(f.t, err)
+	}
+}
+
+// events returns all stored events for a single channel.
+func (f *pruneFixture) events(channelID int64) []*ChannelEvent {
+	f.t.Helper()
+
+	events, err := f.store.GetChannelEvents(
+		f.ctx, channelID, 0, time.Unix(0, 0), f.now.Add(time.Hour),
+		1000,
+	)
+	require.NoError(f.t, err)
+
+	return events
+}
+
+// count returns the total number of events across both channels.
+func (f *pruneFixture) count() int {
+	return len(f.events(f.chan1)) + len(f.events(f.chan2))
+}
+
+// requireAllRecent asserts that every surviving event lies within the
+// retention window, confirming age-based pruning drops the old events rather
+// than the recent ones.
+func (f *pruneFixture) requireAllRecent() {
+	f.t.Helper()
+
+	all := append(f.events(f.chan1), f.events(f.chan2)...)
+	for _, e := range all {
+		require.Equal(f.t, f.recent.Unix(), e.Timestamp.Unix())
+	}
+}
+
+// TestPruneEvents verifies that PruneEvents enforces the max-events count and
+// the retention window independently. Each case runs against its own fresh
+// store so the size and age limits can be exercised in isolation.
+func TestPruneEvents(t *testing.T) {
+	t.Parallel()
+
+	const retention = 30 * 24 * time.Hour
+
+	tests := []struct {
+		name      string
+		seed      func(f *pruneFixture)
+		maxEvents uint64
+		retention time.Duration
+		wantTotal int
+		verify    func(f *pruneFixture)
+	}{{
+		// Pruning an empty table succeeds and deletes nothing.
+		name:      "empty database",
+		maxEvents: 10,
+		retention: retention,
+		wantTotal: 0,
+	}, {
+		// A count equal to max-events is at the ceiling, not over it,
+		// so all events are kept.
+		name: "count equal to max-events keeps all",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan1, f.recent, 5)
+		},
+		maxEvents: 5,
+		retention: retention,
+		wantTotal: 5,
+	}, {
+		// Both limits zero disables pruning entirely, even for events
+		// outside the retention window.
+		name: "both limits zero disables pruning",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan1, f.recent, 5)
+			f.addEvents(f.chan1, f.old, 3)
+		},
+		maxEvents: 0,
+		retention: 0,
+		wantTotal: 8,
+	}, {
+		// The age limit alone drops events older than the window and
+		// keeps the recent ones.
+		name: "age limit prunes old events",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan1, f.recent, 5)
+			f.addEvents(f.chan1, f.old, 3)
+		},
+		maxEvents: 0,
+		retention: retention,
+		wantTotal: 5,
+		verify: func(f *pruneFixture) {
+			f.requireAllRecent()
+		},
+	}, {
+		// The size limit bounds the global table across channels and
+		// keeps the newest events, even when all are inside the
+		// retention window. Channel 2 is seeded last, so its events
+		// have the newest ids and must be the survivors.
+		name: "size limit prunes oldest across channels",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan1, f.recent, 5)
+			f.addEvents(f.chan2, f.recent, 5)
+		},
+		maxEvents: 4,
+		retention: retention,
+		wantTotal: 4,
+		verify: func(f *pruneFixture) {
+			require.Empty(f.t, f.events(f.chan1))
+			require.Len(f.t, f.events(f.chan2), 4)
+		},
+	}, {
+		// With the size ceiling not exceeded, the age limit still
+		// prunes old events independently.
+		name: "age limit prunes with size headroom",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan2, f.recent, 4)
+			f.addEvents(f.chan1, f.old, 3)
+		},
+		maxEvents: 10,
+		retention: retention,
+		wantTotal: 4,
+		verify: func(f *pruneFixture) {
+			f.requireAllRecent()
+		},
+	}, {
+		// Retention zero disables the age limit. With the count under
+		// max-events nothing is pruned.
+		name: "retention zero disables age limit",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan2, f.recent, 4)
+		},
+		maxEvents: 10,
+		retention: 0,
+		wantTotal: 4,
+	}, {
+		// Max-events zero disables the size limit. The age limit still
+		// prunes old events on its own.
+		name: "max-events zero leaves age limit active",
+		seed: func(f *pruneFixture) {
+			f.addEvents(f.chan2, f.recent, 4)
+			f.addEvents(f.chan1, f.old, 3)
+		},
+		maxEvents: 0,
+		retention: retention,
+		wantTotal: 4,
+		verify: func(f *pruneFixture) {
+			f.requireAllRecent()
+		},
+	}}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			f := newPruneFixture(t)
+			if tc.seed != nil {
+				tc.seed(f)
+			}
+
+			_, err := f.store.PruneEvents(
+				f.ctx, tc.maxEvents, tc.retention,
+			)
+			require.NoError(t, err)
+			require.Equal(t, tc.wantTotal, f.count())
+
+			if tc.verify != nil {
+				tc.verify(f)
+			}
+		})
+	}
+}
+
 // TestPagination verifies that the keyset cursor advances correctly across
 // events sharing one second-resolution timestamp.
 func TestPagination(t *testing.T) {

config.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"math"
 	"os"
 	"path"
 	"path/filepath"
@@ -50,6 +51,18 @@ const (
 	// defaultSqliteDatabaseFileName is the default name of the SQLite
 	// database file.
 	defaultSqliteDatabaseFileName = "faraday.db"
+
+	// defaultChanEventsMaxEvents is the default maximum number of channel
+	// events to retain. At roughly 140 bytes per event this acts as a hard
+	// ceiling of approximately 1 GB. A value of 0 disables the size-based
+	// limit.
+	defaultChanEventsMaxEvents = 7000000
+
+	// defaultChanEventsRetention is the default retention window for channel
+	// events. Age-based pruning is disabled by default (0): out of the box
+	// only the max-events size ceiling bounds the table, and operators opt
+	// into a retention window explicitly.
+	defaultChanEventsRetention = 0
 )
 
 var (
@@ -184,6 +197,10 @@ type Config struct { //nolint:maligned
 
 	// Postgres holds the configuration options for a Postgres database
 	Postgres *sqldb.PostgresConfig `group:"postgres" namespace:"postgres"`
+
+	// ChanEvents holds the configuration options for channel event safety
+	// pruning.
+	ChanEvents *chanevents.Config `group:"chanevents" namespace:"chanevents"`
 }
 
 // DefaultConfig returns all default values for the Config struct.
@@ -209,6 +226,10 @@ func DefaultConfig() Config {
 		Sqlite: &db.SqliteConfig{
 			DatabaseFileName: defaultSqliteDatabaseFileName,
 		},
+		ChanEvents: &chanevents.Config{
+			MaxEvents: defaultChanEventsMaxEvents,
+			Retention: defaultChanEventsRetention,
+		},
 	}
 }
 
@@ -343,6 +364,25 @@ func ValidateConfig(config *Config) error {
 		config.Lnd.TLSCertPath,
 	)
 
+	if config.ChanEvents != nil {
+		// The channel event size limit becomes an int32 SQL OFFSET
+		// during pruning, so reject values that would overflow it and
+		// silently corrupt the prune bound.
+		if config.ChanEvents.MaxEvents > math.MaxInt32 {
+			return fmt.Errorf("chanevents.max-events must not "+
+				"exceed %d", math.MaxInt32)
+		}
+
+		// A negative retention is silently ignored by the prune checks,
+		// which only treat a strictly positive duration as enabling
+		// age-based pruning. Reject it so a misconfigured window fails
+		// loudly instead of disabling pruning unexpectedly.
+		if config.ChanEvents.Retention < 0 {
+			return fmt.Errorf("chanevents.retention must not be " +
+				"negative")
+		}
+	}
+
 	return nil
 }
 

faraday.go

@@ -580,9 +580,17 @@ func (f *Faraday) initialize(withMacaroonService bool) error {
 		return fmt.Errorf("could not create stores: %v", err)
 	}
 
-	// Create the channel event monitor.
+	// Create the channel event monitor. ChanEvents may be nil on
+	// initialization paths that don't go through DefaultConfig (e.g. when
+	// faraday runs as a subserver), so fall back to a zero-value config
+	// instead of dereferencing a nil pointer.
+	var chanEventsCfg chanevents.Config
+	if f.cfg.ChanEvents != nil {
+		chanEventsCfg = *f.cfg.ChanEvents
+	}
+
 	f.monitor = chanevents.NewMonitor(
-		f.lnd.Client, f.stores.ChanEventsStore,
+		f.lnd.Client, f.stores.ChanEventsStore, chanEventsCfg,
 	)
 
 	ctx, cancel := context.WithCancel(context.Background())