From e852394bddf007e9169939def962537a1a852343 Mon Sep 17 00:00:00 2001 From: bitromortac Date: Mon, 22 Jun 2026 10:10:00 +0200 Subject: [PATCH 1/4] db: add channel event pruning query and index Add a PruneChannelEvents query that bounds the channel_events table by both size and age in a single statement: an id-keyset offset enforces a maximum event count and a timestamp filter enforces a retention window, OR-joined so each limit applies independently. The query returns the number of rows deleted so callers can surface pruning activity. Add a standalone timestamp index so the global age-based prune does not scan the full table. The existing composite index leads with channel_id and cannot serve a channel-agnostic timestamp filter. --- db/migrations.go | 2 +- db/sqlc/chanevents.sql.go | 38 +++++++++++++++++++ .../000002_chanevents_ts_idx.down.sql | 1 + .../000002_chanevents_ts_idx.up.sql | 6 +++ db/sqlc/querier.go | 11 ++++++ db/sqlc/queries/chanevents.sql | 20 ++++++++++ 6 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 db/sqlc/migrations/000002_chanevents_ts_idx.down.sql create mode 100644 db/sqlc/migrations/000002_chanevents_ts_idx.up.sql diff --git a/db/migrations.go b/db/migrations.go index bd882c0..6520dd3 100644 --- a/db/migrations.go +++ b/db/migrations.go @@ -6,5 +6,5 @@ const ( // daemon. // // NOTE: This MUST be updated when a new migration is added. - LatestMigrationVersion = 1 + LatestMigrationVersion = 2 ) diff --git a/db/sqlc/chanevents.sql.go b/db/sqlc/chanevents.sql.go index ba1dee2..77a05c7 100644 --- a/db/sqlc/chanevents.sql.go +++ b/db/sqlc/chanevents.sql.go @@ -227,3 +227,41 @@ func (q *Queries) InsertPeer(ctx context.Context, pubkey string) (int64, error) err := row.Scan(&id) return id, err } + +const pruneChannelEventsByAge = `-- name: PruneChannelEventsByAge :execrows +DELETE FROM channel_events +WHERE channel_events.timestamp < $1 +` + +// PruneChannelEventsByAge enforces the retention window on the channel_events +// table, returning the number of rows deleted. It deletes any row whose +// timestamp predates the given cutoff. +func (q *Queries) PruneChannelEventsByAge(ctx context.Context, timestamp time.Time) (int64, error) { + result, err := q.db.ExecContext(ctx, pruneChannelEventsByAge, timestamp) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +const pruneChannelEventsBySize = `-- name: PruneChannelEventsBySize :execrows +DELETE FROM channel_events +WHERE channel_events.id < COALESCE(( + SELECT id FROM channel_events + ORDER BY id DESC + LIMIT 1 OFFSET $1 +), 0) +` + +// PruneChannelEventsBySize enforces the size ceiling on the channel_events +// table, returning the number of rows deleted. It keeps the newest rows by +// deleting everything with a smaller (earlier-inserted) id than the id found at +// the given offset from the newest row, so an offset of (max-events - 1) keeps +// exactly max-events rows. +func (q *Queries) PruneChannelEventsBySize(ctx context.Context, offset int32) (int64, error) { + result, err := q.db.ExecContext(ctx, pruneChannelEventsBySize, offset) + if err != nil { + return 0, err + } + return result.RowsAffected() +} diff --git a/db/sqlc/migrations/000002_chanevents_ts_idx.down.sql b/db/sqlc/migrations/000002_chanevents_ts_idx.down.sql new file mode 100644 index 0000000..0c18632 --- /dev/null +++ b/db/sqlc/migrations/000002_chanevents_ts_idx.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS channel_events_ts_idx; diff --git a/db/sqlc/migrations/000002_chanevents_ts_idx.up.sql b/db/sqlc/migrations/000002_chanevents_ts_idx.up.sql new file mode 100644 index 0000000..9f3aa6c --- /dev/null +++ b/db/sqlc/migrations/000002_chanevents_ts_idx.up.sql @@ -0,0 +1,6 @@ +-- This standalone timestamp index supports the global age-based prune in +-- PruneChannelEvents, which deletes across all channels by timestamp with no +-- channel_id predicate. The composite (channel_id, timestamp) index cannot +-- serve that query because its leading column is channel_id, so without this +-- index every retention prune would scan the full channel_events table. +CREATE INDEX IF NOT EXISTS channel_events_ts_idx ON channel_events (timestamp); diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index b62ec68..5870af6 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -6,6 +6,7 @@ package sqlc import ( "context" + "time" ) type Querier interface { @@ -18,6 +19,16 @@ type Querier interface { InsertChannel(ctx context.Context, arg InsertChannelParams) (int64, error) InsertChannelEvent(ctx context.Context, arg InsertChannelEventParams) error InsertPeer(ctx context.Context, pubkey string) (int64, error) + // PruneChannelEventsByAge enforces the retention window on the channel_events + // table, returning the number of rows deleted. It deletes any row whose + // timestamp predates the given cutoff. + PruneChannelEventsByAge(ctx context.Context, timestamp time.Time) (int64, error) + // PruneChannelEventsBySize enforces the size ceiling on the channel_events + // table, returning the number of rows deleted. It keeps the newest rows by + // deleting everything with a smaller (earlier-inserted) id than the id found at + // the given offset from the newest row, so an offset of (max-events - 1) keeps + // exactly max-events rows. + PruneChannelEventsBySize(ctx context.Context, offset int32) (int64, error) } var _ Querier = (*Queries)(nil) diff --git a/db/sqlc/queries/chanevents.sql b/db/sqlc/queries/chanevents.sql index d9d1629..d32498d 100644 --- a/db/sqlc/queries/chanevents.sql +++ b/db/sqlc/queries/chanevents.sql @@ -38,3 +38,23 @@ LIMIT 1; SELECT c.id, c.short_channel_id, p.pubkey FROM channels c JOIN peers p ON c.peer_id = p.id; + +-- name: PruneChannelEventsBySize :execrows +-- PruneChannelEventsBySize enforces the size ceiling on the channel_events +-- table, returning the number of rows deleted. It keeps the newest rows by +-- deleting everything with a smaller (earlier-inserted) id than the id found at +-- the given offset from the newest row, so an offset of (max-events - 1) keeps +-- exactly max-events rows. +DELETE FROM channel_events +WHERE channel_events.id < COALESCE(( + SELECT id FROM channel_events + ORDER BY id DESC + LIMIT 1 OFFSET $1 +), 0); + +-- name: PruneChannelEventsByAge :execrows +-- PruneChannelEventsByAge enforces the retention window on the channel_events +-- table, returning the number of rows deleted. It deletes any row whose +-- timestamp predates the given cutoff. +DELETE FROM channel_events +WHERE channel_events.timestamp < $1; From dccbf53a17989fed8d412f3a2f60796c4bfbcaf9 Mon Sep 17 00:00:00 2001 From: bitromortac Date: Mon, 22 Jun 2026 10:10:00 +0200 Subject: [PATCH 2/4] 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 | 16 +++ chanevents/monitor.go | 89 ++++++++++++++- chanevents/store.go | 61 +++++++++++ chanevents/store_test.go | 231 +++++++++++++++++++++++++++++++++++++++ config.go | 40 +++++++ faraday.go | 12 +- 6 files changed, 444 insertions(+), 5 deletions(-) diff --git a/chanevents/chanevents.go b/chanevents/chanevents.go index 6359195..93d08fd 100644 --- a/chanevents/chanevents.go +++ b/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 diff --git a/chanevents/monitor.go b/chanevents/monitor.go index 9e1c93e..1d9fa41 100644 --- a/chanevents/monitor.go +++ b/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 " + diff --git a/chanevents/store.go b/chanevents/store.go index b454a33..1b13b3a 100644 --- a/chanevents/store.go +++ b/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] diff --git a/chanevents/store_test.go b/chanevents/store_test.go index 232de48..bb5a49c 100644 --- a/chanevents/store_test.go +++ b/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) { diff --git a/config.go b/config.go index 70713ee..c8002c4 100644 --- a/config.go +++ b/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 } diff --git a/faraday.go b/faraday.go index 23b5255..c1e63cf 100644 --- a/faraday.go +++ b/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()) From c5d7147ab0bc661085e7668ce1fd577404b27e3d Mon Sep 17 00:00:00 2001 From: bitromortac Date: Mon, 22 Jun 2026 10:10:00 +0200 Subject: [PATCH 3/4] itest: add channel events pruning test Add an integration test to verify the safety pruning behavior under various conditions. --- itest/channel_events_test.go | 73 ++++++++++++++++++++++++++++++++++++ itest/test_context.go | 11 ++++-- 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/itest/channel_events_test.go b/itest/channel_events_test.go index 4def9f6..b1befbd 100644 --- a/itest/channel_events_test.go +++ b/itest/channel_events_test.go @@ -360,3 +360,76 @@ func TestForwardingDowntime(t *testing.T) { return ok }, "expected bob self-pair after reconnect") } + +// TestChannelEventsPruning verifies that starting Faraday with low size limits +// (e.g. max-events=1 and retention=2s) executes live background pruning +// successfully and bounds the database size correctly. +func TestChannelEventsPruning(t *testing.T) { + c := newTestContext( + t, "--chanevents.max-events=1", "--chanevents.retention=2s", + ) + defer c.stop() + + ctx := context.Background() + + // We will start by opening a channel from alice to bob. + var aliceChannelAmt = btcutil.Amount(500000) + + err := c.aliceClient.Client.Connect( + ctx, c.bobPubkey, "localhost:10012", true, + ) + require.NoError(c.t, err, "could not connect nodes") + + aliceChannel, _ := c.openChannel( + c.aliceClient.Client, c.bobPubkey, aliceChannelAmt, + ) + + // Use a far-future end time so the query window never excludes a stored + // event on a slow host. A tight wall-clock window here would make the + // counts below racy. + endTime := time.Now().Add(time.Hour).Unix() + + // We deliberately do not assert on the initial event count here: opening + // a channel records several events, but the 2-second background prune can + // fire before we observe them on a slow host, so any such pre-prune + // assertion would be flaky. The eventuallyf checks below verify the + // pruning behaviour directly instead. + + // Wait for the live background pruning ticker to bound the table to the + // max-events ceiling. We assert at most one event rather than exactly + // one: the size limit keeps a single event, but the 2-second retention + // limit then ages it out since no new events follow the channel open, + // so the steady state is zero or one. + var eventsAfter *frdrpc.ChannelEventsResponse + c.eventuallyf(func() bool { + var err error + eventsAfter, err = c.faradayClient.GetChannelEvents( + ctx, &frdrpc.ChannelEventsRequest{ + ChanPoint: aliceChannel.String(), + EndTime: endTime, + }, + ) + if err != nil { + return false + } + return len(eventsAfter.Events) <= 1 + }, "expected channel events to be pruned down to at most one in the "+ + "background") + + // No further events follow the channel open, so once the remaining + // event ages past the 2-second retention window the age-based prune + // removes it too, draining the table to zero. + c.eventuallyf(func() bool { + eventsAfter, err := c.faradayClient.GetChannelEvents( + ctx, &frdrpc.ChannelEventsRequest{ + ChanPoint: aliceChannel.String(), + EndTime: endTime, + }, + ) + if err != nil { + return false + } + return len(eventsAfter.Events) == 0 + }, "expected channel events to be pruned down to zero once all events "+ + "age out of the retention window") +} diff --git a/itest/test_context.go b/itest/test_context.go index 70d4e32..2f34a2f 100644 --- a/itest/test_context.go +++ b/itest/test_context.go @@ -68,7 +68,7 @@ type testContext struct { } // newTestContext returns a new context instance. -func newTestContext(t *testing.T) *testContext { +func newTestContext(t *testing.T, extraFaradayArgs ...string) *testContext { var err error ctx := &testContext{ @@ -123,7 +123,7 @@ func newTestContext(t *testing.T) *testContext { require.NoError(t, err) // Start faraday. - ctx.startFaraday() + ctx.startFaraday(extraFaradayArgs...) // Wait for faraday's channel events monitor to finish its initial // chain-sync. @@ -564,10 +564,13 @@ func (c *testContext) waitForMempoolTxCount(txCount int, msg string) { // startFaraday starts faraday, connecting to our test context's alice lnd node. // It returns process start errors and an error channel for errors that occur // after the start. -func (c *testContext) startFaraday() { +func (c *testContext) startFaraday(extraArgs ...string) { + args := append([]string{}, faradayArgs...) + args = append(args, extraArgs...) + // Start loop client daemon. c.faradayCmd = exec.Command( - faradayCmd, faradayArgs..., + faradayCmd, args..., ) attachPrefixStdout(c.faradayCmd, "faraday") From 754d66078652ee36021ea5a81bcf18e23e41b971 Mon Sep 17 00:00:00 2001 From: bitromortac Date: Mon, 22 Jun 2026 10:10:00 +0200 Subject: [PATCH 4/4] docs: document channel event safety pruning flags Document the new configuration flags in the README, providing rough storage estimation metrics (~100 bytes per event) to guide database sizing. --- README.md | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/README.md b/README.md index 0bca937..d6dd807 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,42 @@ Faraday serves requests over grpc by default on `localhost:8465`. This default c --rpclisten={host:port to listen for requests} ``` +#### Channel Event Storage +Faraday records channel events (online/offline transitions and balance updates) +in its database. On high-frequency channels this table can grow without bound, +so a size ceiling is enabled by default, with an optional age-based retention +window: +```text +--chanevents.max-events={maximum number of events to retain} +--chanevents.retention={minimum duration of events to keep, e.g. 1440h} +``` +By default only the size ceiling is active: a maximum of 7 million events +(`--chanevents.max-events=7000000`, roughly 1 GB of storage). Age-based +retention is disabled by default (`--chanevents.retention=0`) so that history is +never aged out unless an operator opts in. The retention window is a Go duration +string (e.g. `1440h` for 60 days). The two limits are applied independently: + +1. **Size Ceiling (Hard Limit):** If the database exceeds `max-events`, older + events are pruned unconditionally to ensure the database size is strictly + capped, preventing disk filling. Newer events inside the retention window can + still be pruned if needed to satisfy this size limit. +2. **Age Threshold (Freshness):** When `retention` is set to a non-zero + duration, any events older than that window are automatically pruned to keep + history fresh, even if the database size is below `max-events`. + +Because the two limits are independent, disabling pruning entirely requires +turning off both: `--chanevents.max-events=0 --chanevents.retention=0`. Setting +only `max-events=0` disables the size ceiling, leaving the table bounded only by +any retention window that has been configured. + +As a rough rule of thumb, each channel event consumes on the order of 140 bytes +of storage once table and index overhead is taken into account. The default +`--chanevents.max-events=7000000` therefore bounds the table at roughly 1 GB +(7 million events × ~140 bytes ≈ 1 GB). For roughly 100 MB use +`--chanevents.max-events=700000`. These are approximations measured on a +compacted SQLite database, and actual usage runs higher on a live database +(write-ahead log, page fragmentation) and varies by backend. + #### Cli Tool The RPC server can be conveniently accessed using a command line tool. 1. Run faraday as detailed above