From 772549516d09ecf63c291edf649de5d60a18043c Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Thu, 3 Sep 2026 20:07:41 +0700 Subject: [PATCH] feat: prune repeated stream values on a schedule --- extensions/tn_digest/README.md | 57 +++- extensions/tn_digest/constants.go | 4 + extensions/tn_digest/extension.go | 87 +++++- extensions/tn_digest/leader_reload_test.go | 14 + extensions/tn_digest/prune_scheduler_test.go | 213 +++++++++++++ extensions/tn_digest/scheduler/constants.go | 31 ++ .../tn_digest/scheduler/drain_slot_test.go | 163 ++++++++++ extensions/tn_digest/scheduler/scheduler.go | 291 +++++++++++++++++- extensions/tn_digest/scheduler_lifecycle.go | 10 + extensions/tn_digest/tn_digest.go | 42 ++- 10 files changed, 898 insertions(+), 14 deletions(-) create mode 100644 extensions/tn_digest/prune_scheduler_test.go create mode 100644 extensions/tn_digest/scheduler/drain_slot_test.go diff --git a/extensions/tn_digest/README.md b/extensions/tn_digest/README.md index cbec5a79e..48eae46ab 100644 --- a/extensions/tn_digest/README.md +++ b/extensions/tn_digest/README.md @@ -2,10 +2,17 @@ ## What it does (brief) - Periodically calls a Kuneiform action `main.auto_digest()` via real, signed transactions. +- Periodically calls `main.auto_prune_duplicates()` the same way, on its own schedule. - Runs only when this node is the block leader (leader-gated scheduler with consensus checks). -- Reads enable/schedule from on-chain table `digest_config` and reconciles changes every N blocks. +- Reads enable/schedule from on-chain tables `digest_config` and `duplicate_prune_config`, and reconciles changes every N blocks. - Uses singleton scheduler to prevent overlapping jobs and supports both 5-field and 6-field cron expressions. +The two jobs are independent. Separate tables, separate enable flags, separate +schedules, separate crons, so turning one off or rescheduling it leaves the other +alone. They do share one thing: only one of them drains at a time, because both +broadcast from the node's signer account and two drains in flight would take the +same nonce. + --- ## Operators @@ -35,9 +42,53 @@ VALUES (1, true, '0 9 * * *'); UPDATE main.digest_config SET enabled = true, digest_schedule = '*/10 * * * *' WHERE id = 1; ``` +### Duplicate pruning + +A stream that publishes the same value every day stores one row a day forever, and +reads carry the last observation forward, so every row after the first in such a run +answers exactly what the row before it already answered. `auto_prune_duplicates` +removes them. Migrations 056 and 057 hold the rule and the reasoning; the summary is +that no read changes its answer, but a range read returns fewer points and an +anchored read reports an older `event_time` beside the same value. + +The extension reads a single row in `duplicate_prune_config` (id = 1): +- `enabled` (boolean): turns the sweep on and off. **Ships false**, and it is the + only gate, and there is no build flag to set as well. +- `prune_schedule` (cron string): when the sweep runs. Defaults to `0 */6 * * *`. +- `retention_days` (int): how old a record must be before it is a candidate. + Defaults to 30. The scheduler does not pass this, so changing it here changes + what the sweep does without a release. +- `last_stream_ref` (int): where the sweep is. Duplicate-ness is a property of a + whole stream, so there is no queue to drain: the sweep walks `streams.id` in + order and wraps at the end. + +Unlike `digest_config`, migration 056 seeds this row, so a network that has run its +migrations always has one. + +```sql +-- Turn the sweep on +UPDATE main.duplicate_prune_config SET enabled = true WHERE id = 1; + +-- Prune less aggressively, or reschedule +UPDATE main.duplicate_prune_config +SET retention_days = 90, prune_schedule = '0 3 * * *' WHERE id = 1; +``` + +Both tables are consensus state, so change them through a signed +`kwil-cli exec-sql` rather than psql: a direct write on one node diverges its +AppHash. + +**Before turning it on**, read the two things a firing costs. Each run visits 100 +streams and scans their whole history inside one consensus transaction, and a run +that deletes leaves dead tuples behind, so pruning a long backlog wants a +`pg_repack` after it, with the transient disk that implies. And the sweep is cyclic, so +`has_more_to_delete` means "the cursor has not finished a pass" rather than "there +is more to delete": a firing runs its whole loop rather than stopping early, which +on a large network is by design. + ### Leader Gating & Lifecycle -- Scheduler starts only when this node becomes leader and `enabled = true`. -- Scheduler stops immediately when leadership is lost or when `enabled` becomes false. +- Each job starts only when this node becomes leader and its own `enabled = true`. +- Both stop immediately when leadership is lost, and each stops when its own `enabled` becomes false. - The extension checks the config again every N blocks (default 1000, configurable below). ### Configuration (TOML) diff --git a/extensions/tn_digest/constants.go b/extensions/tn_digest/constants.go index 16ab7fb64..4520cee12 100644 --- a/extensions/tn_digest/constants.go +++ b/extensions/tn_digest/constants.go @@ -3,4 +3,8 @@ package tn_digest const ( ExtensionName = "tn_digest" DefaultDigestSchedule = "0 */6 * * *" // every 6 hours + + // DefaultPruneSchedule matches duplicate_prune_config's own default, so it only + // ever applies on a network whose row or column is missing. + DefaultPruneSchedule = "0 */6 * * *" // every 6 hours ) diff --git a/extensions/tn_digest/extension.go b/extensions/tn_digest/extension.go index 5736fec05..9a9c71771 100644 --- a/extensions/tn_digest/extension.go +++ b/extensions/tn_digest/extension.go @@ -35,6 +35,11 @@ type Extension struct { enabled bool schedule string + // duplicate prune config snapshot, read from duplicate_prune_config rather + // than digest_config and gating a separate cron + pruneEnabled bool + pruneSchedule string + // reload policy reloadIntervalBlocks int64 lastCheckedHeight int64 @@ -94,11 +99,23 @@ func (e *Extension) SetConfig(enabled bool, schedule string) { } func (e *Extension) ConfigEnabled() bool { return e.enabled } func (e *Extension) Schedule() string { return e.schedule } +func (e *Extension) SetPruneConfig(enabled bool, schedule string) { + e.pruneEnabled = enabled + e.pruneSchedule = schedule +} +func (e *Extension) PruneEnabled() bool { return e.pruneEnabled } +func (e *Extension) PruneSchedule() string { + if e.pruneSchedule == "" { + return DefaultPruneSchedule + } + return e.pruneSchedule +} func (e *Extension) SetScheduler(s *scheduler.DigestScheduler) { if e.scheduler == s { return } if e.scheduler != nil { + _ = e.scheduler.StopPrune() _ = e.scheduler.Stop() } e.scheduler = s @@ -183,14 +200,27 @@ func (e *Extension) retryConfigReload() { } enabled, schedule, err := e.EngineOps().LoadDigestConfig(e.retryWorkerCtx) + var pruneEnabled bool + var pruneSchedule string + var pruneErr error if err == nil { + // Both configs are reloaded together so one worker covers both crons. + pruneEnabled, pruneSchedule, pruneErr = e.EngineOps().LoadPruneConfig(e.retryWorkerCtx) + } + if err == nil && pruneErr == nil { // Success! Update config (app=nil since we're in background, service already cached) - e.Logger().Info("config reload succeeded in background", "attempt", attempt, "enabled", enabled, "schedule", schedule) + e.Logger().Info("config reload succeeded in background", "attempt", attempt, + "enabled", enabled, "schedule", schedule, + "prune_enabled", pruneEnabled, "prune_schedule", pruneSchedule) e.applyConfigChangeWithLock(e.retryWorkerCtx, enabled, schedule, nil) + e.applyPruneConfigChangeWithLock(e.retryWorkerCtx, pruneEnabled, pruneSchedule, nil) return } + if err == nil { + err = pruneErr + } - // Check if context was cancelled during LoadDigestConfig + // Check if context was cancelled during the reload if e.retryWorkerCtx.Err() != nil { e.Logger().Info("retry worker cancelled during config reload") return @@ -247,10 +277,63 @@ func (e *Extension) applyConfigChangeWithLock(ctx context.Context, enabled bool, } } +// applyPruneConfigChangeWithLock applies duplicate_prune_config changes with the +// same synchronization applyConfigChangeWithLock uses, and shares its lock so a +// single reload cannot have the two crons half-applied. +func (e *Extension) applyPruneConfigChangeWithLock(ctx context.Context, enabled bool, schedule string, app *common.App) { + e.retryMu.Lock() + defer e.retryMu.Unlock() + + if schedule == "" { + schedule = DefaultPruneSchedule + } + + if enabled == e.PruneEnabled() && schedule == e.PruneSchedule() { + return + } + + e.Logger().Info("duplicate prune config changed, updating scheduler", + "old_enabled", e.PruneEnabled(), + "new_enabled", enabled, + "old_schedule", e.PruneSchedule(), + "new_schedule", schedule, + "is_leader", e.IsLeader()) + e.SetPruneConfig(enabled, schedule) + + if !enabled { + e.stopPruneIfRunning() + e.Logger().Info("duplicate prune stopped due to config disabled") + return + } + if !e.IsLeader() { + e.Logger().Info("duplicate prune config enabled but not leader, will start when leadership acquired") + return + } + + service := e.Service() + if app != nil && app.Service != nil { + service = app.Service + if e.Service() == nil { + e.SetService(service) + } + } + if e.Scheduler() == nil && !e.ensureSchedulerWithService(service) { + e.Logger().Debug("tn_digest: prerequisites missing; deferring duplicate prune (re)start after config update") + return + } + e.stopPruneIfRunning() + if err := e.startPruneScheduler(ctx); err != nil { + e.Logger().Warn("failed to (re)start duplicate prune scheduler after config update", "error", err) + } else { + e.Logger().Info("duplicate prune (re)started with new schedule", "schedule", e.PruneSchedule()) + } +} + // Close stops background jobs. func (e *Extension) Close() { e.stopRetryWorker() if e.scheduler != nil { + _ = e.scheduler.StopPrune() _ = e.scheduler.Stop() } } diff --git a/extensions/tn_digest/leader_reload_test.go b/extensions/tn_digest/leader_reload_test.go index 9c8329ae0..aa88dc422 100644 --- a/extensions/tn_digest/leader_reload_test.go +++ b/extensions/tn_digest/leader_reload_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "math/big" + "strings" "testing" "time" @@ -39,6 +40,12 @@ func (p testPubKey) Verify(data []byte, sig []byte) (bool, error) { return true, type fakeDB struct { enabled bool schedule string + // duplicate_prune_config is a separate row in a separate table, and the two + // features move independently, so it answers from its own fields. Leaving + // pruneSchedule empty answers no row, which is a network that has not been + // migrated as far as 056. + pruneEnabled bool + pruneSchedule string // For testing transient failures failCount int // number of times to fail before succeeding callCount int // current call count @@ -55,6 +62,13 @@ func (f *fakeDB) Execute(ctx context.Context, stmt string, args ...any) (*sqltyp return nil, errors.New("database timeout") } + if strings.Contains(stmt, "duplicate_prune_config") { + if f.pruneSchedule == "" { + return &sqltypes.ResultSet{Columns: []string{"enabled", "prune_schedule"}, Rows: [][]any{}}, nil + } + return &sqltypes.ResultSet{Columns: []string{"enabled", "prune_schedule"}, Rows: [][]any{{f.pruneEnabled, f.pruneSchedule}}}, nil + } + // Return one row for SELECT enabled, digest_schedule FROM digest_config WHERE id = 1 // Any other stmt returns empty rows if len(stmt) >= 6 && stmt[:6] == "SELECT" { diff --git a/extensions/tn_digest/prune_scheduler_test.go b/extensions/tn_digest/prune_scheduler_test.go new file mode 100644 index 000000000..6894e4499 --- /dev/null +++ b/extensions/tn_digest/prune_scheduler_test.go @@ -0,0 +1,213 @@ +package tn_digest + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/trufnetwork/kwil-db/common" + "github.com/trufnetwork/kwil-db/core/log" + digestinternal "github.com/trufnetwork/node/extensions/tn_digest/internal" +) + +// The duplicate prune sweep shares the extension with digest but nothing else: +// its own table, its own enabled flag, its own schedule and its own cron. These +// tests are mostly about that separation, because the failure it protects against +// is one feature's config change silently stopping the other. + +// Nothing prunes until an operator turns duplicate_prune_config.enabled on, and +// the migration ships it false. With digest off as well the extension builds no +// scheduler at all. +func TestPrune_DefaultDisabled_NoSchedulerOnLeaderAcquire(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(false, "*/5 * * * *") + ext.SetPruneConfig(false, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1000) + identity := []byte("pruneA") + app := &common.App{Service: makeService(identity, "1000")} + ext.SetService(app.Service) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + assert.Nil(t, ext.Scheduler()) +} + +// Pruning does not depend on digest being on. An operator draining duplicates on a +// network where digest is off should get the sweep and nothing else. +func TestPrune_LeaderAcquire_StartsPruneWithDigestOff(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(false, "*/5 * * * *") + ext.SetPruneConfig(true, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1000) + identity := []byte("pruneB") + app := &common.App{Service: makeService(identity, "1000")} + ext.SetService(app.Service) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.NotNil(t, ext.Scheduler()) + assert.True(t, ext.Scheduler().PruneRunning()) + assert.False(t, ext.Scheduler().Running()) + + _ = ext.Scheduler().StopPrune() +} + +func TestPrune_LoseLeadership_StopsPrune(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(false, "*/5 * * * *") + ext.SetPruneConfig(true, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1000) + identity := []byte("pruneC") + app := &common.App{Service: makeService(identity, "1000")} + ext.SetService(app.Service) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.NotNil(t, ext.Scheduler()) + require.True(t, ext.Scheduler().PruneRunning()) + + digestLeaderLose(context.Background(), app, makeBlock(2, []byte("someone else"))) + assert.False(t, ext.Scheduler().PruneRunning()) +} + +// The enable path an operator actually takes: set enabled through a signed +// exec-sql and wait for the next config reload to pick it up. +func TestPrune_Reload_EnablesAndStarts_WhenBecomesEnabled(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(false, "*/5 * * * *") + ext.SetPruneConfig(false, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1) + ext.SetLastCheckedHeight(1) + identity := []byte("pruneD") + app := &common.App{Service: makeService(identity, "1")} + ext.SetService(app.Service) + + fdb := &fakeDB{pruneEnabled: true, pruneSchedule: "*/5 * * * *"} + ext.SetEngineOps(digestinternal.NewEngineOperations(&fakeEngine{}, fdb, nil, &fakeAccounts{}, log.New())) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.Nil(t, ext.Scheduler()) + + digestLeaderEndBlock(context.Background(), app, makeBlock(2, identity)) + require.NotNil(t, ext.Scheduler()) + assert.True(t, ext.Scheduler().PruneRunning()) + assert.False(t, ext.Scheduler().Running(), "digest is off and should have stayed off") + + _ = ext.Scheduler().StopPrune() +} + +// The way back is the same knob. Setting enabled false stops the sweep without a +// binary release, which is the reason the gate lives in the table rather than in a +// Go constant. +func TestPrune_Reload_DisablesAndStops_WhenBecomesDisabled(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(false, "*/5 * * * *") + ext.SetPruneConfig(true, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1) + ext.SetLastCheckedHeight(1) + identity := []byte("pruneE") + app := &common.App{Service: makeService(identity, "1")} + ext.SetService(app.Service) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.NotNil(t, ext.Scheduler()) + require.True(t, ext.Scheduler().PruneRunning()) + + fdb := &fakeDB{pruneEnabled: false, pruneSchedule: "*/5 * * * *"} + ext.SetEngineOps(digestinternal.NewEngineOperations(&fakeEngine{}, fdb, nil, &fakeAccounts{}, log.New())) + digestLeaderEndBlock(context.Background(), app, makeBlock(2, identity)) + + assert.False(t, ext.Scheduler().PruneRunning()) + assert.False(t, ext.PruneEnabled()) +} + +// The reason the sweep gets its own cron and its own context. A digest schedule +// change stops and restarts the digest cron; on a shared one that would cancel a +// prune drain partway through a six-hour sweep, and there is no signal that would +// tell anyone it had happened. +func TestPrune_SurvivesADigestConfigChange(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(true, "*/5 * * * *") + ext.SetPruneConfig(true, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1) + ext.SetLastCheckedHeight(1) + identity := []byte("pruneF") + app := &common.App{Service: makeService(identity, "1")} + ext.SetService(app.Service) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.NotNil(t, ext.Scheduler()) + require.True(t, ext.Scheduler().Running()) + require.True(t, ext.Scheduler().PruneRunning()) + + // Digest moves to a different schedule; the prune row is unchanged. + fdb := &fakeDB{ + enabled: true, schedule: "0 9 * * *", + pruneEnabled: true, pruneSchedule: "*/5 * * * *", + } + ext.SetEngineOps(digestinternal.NewEngineOperations(&fakeEngine{}, fdb, nil, &fakeAccounts{}, log.New())) + digestLeaderEndBlock(context.Background(), app, makeBlock(2, identity)) + + assert.Equal(t, "0 9 * * *", ext.Schedule()) + assert.True(t, ext.Scheduler().PruneRunning(), "the prune sweep should not notice a digest config change") + + _ = ext.Scheduler().StopPrune() + _ = ext.Scheduler().Stop() +} + +// The other direction of the same separation. Turning digest off stops the digest +// cron; the sweep is a different feature answering to a different row and has to +// keep going. +func TestPrune_SurvivesDigestBeingTurnedOff(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(true, "*/5 * * * *") + ext.SetPruneConfig(true, "*/5 * * * *") + ext.SetReloadIntervalBlocks(1) + ext.SetLastCheckedHeight(1) + identity := []byte("pruneH") + app := &common.App{Service: makeService(identity, "1")} + ext.SetService(app.Service) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.NotNil(t, ext.Scheduler()) + require.True(t, ext.Scheduler().Running()) + require.True(t, ext.Scheduler().PruneRunning()) + + fdb := &fakeDB{ + enabled: false, schedule: "*/5 * * * *", + pruneEnabled: true, pruneSchedule: "*/5 * * * *", + } + ext.SetEngineOps(digestinternal.NewEngineOperations(&fakeEngine{}, fdb, nil, &fakeAccounts{}, log.New())) + digestLeaderEndBlock(context.Background(), app, makeBlock(2, identity)) + + assert.False(t, ext.Scheduler().Running()) + assert.True(t, ext.Scheduler().PruneRunning(), "turning digest off should not stop the sweep") + + _ = ext.Scheduler().StopPrune() +} + +// A node whose binary is ahead of its migrations reads no duplicate_prune_config +// at all. That has to leave the sweep off rather than fail the reload, or every +// end-block on such a node would signal the retry worker. +func TestPrune_MissingConfigLeavesTheSweepOff(t *testing.T) { + ext := resetExtensionForTest() + ext.SetConfig(true, "*/5 * * * *") + ext.SetPruneConfig(false, "") + ext.SetReloadIntervalBlocks(1) + ext.SetLastCheckedHeight(1) + identity := []byte("pruneG") + app := &common.App{Service: makeService(identity, "1")} + ext.SetService(app.Service) + + // pruneSchedule empty means the fake answers no row. + fdb := &fakeDB{enabled: true, schedule: "*/5 * * * *"} + ext.SetEngineOps(digestinternal.NewEngineOperations(&fakeEngine{}, fdb, nil, &fakeAccounts{}, log.New())) + + digestLeaderAcquire(context.Background(), app, makeBlock(1, identity)) + require.NotNil(t, ext.Scheduler()) + digestLeaderEndBlock(context.Background(), app, makeBlock(2, identity)) + + assert.False(t, ext.PruneEnabled()) + assert.False(t, ext.Scheduler().PruneRunning()) + assert.True(t, ext.Scheduler().Running(), "digest should be unaffected") + + _ = ext.Scheduler().Stop() +} diff --git a/extensions/tn_digest/scheduler/constants.go b/extensions/tn_digest/scheduler/constants.go index f258c4953..abcc92eac 100644 --- a/extensions/tn_digest/scheduler/constants.go +++ b/extensions/tn_digest/scheduler/constants.go @@ -29,4 +29,35 @@ const ( // indexer fallback (trufscan #183) is live in prod, so a pruned tx still // resolves on the explorer /tx page. TrimTxEventsEnabled bool = false + + // Duplicate prune constants. + // + // There is no Enabled constant here on purpose. duplicate_prune_config.enabled + // ships false and is the only gate, so an operator turns the sweep on with a + // signed exec-sql rather than a binary release. A second gate in Go would mean + // setting that column and watching nothing happen. + // + // The sweep is cyclic: has_more_to_delete means "the cursor has not finished a + // pass", not "there is more to delete". A firing therefore runs its whole loop + // rather than stopping early, so these numbers say how much of a pass one + // firing covers rather than how fast a backlog drains. + // + // Mainnet holds ~182,000 primitive streams. At 100 streams a run and 100 runs a + // firing that is 10,000 streams, so a pass takes ~19 firings: about five days on + // the six-hourly default. Raising PruneStreamBatchSize shortens that, and the + // cost is a longer scan inside one consensus transaction -- measure with + // internal/benchmark/digest before doing it on mainnet. + PruneDeleteCap = 100_000 + PruneStreamBatchSize = 100 + PruneDrainMaxRuns = 100 + + // PruneDrainRunDelay paces the runs that actually delete, the way digest's + // DrainRunDelay paces its own capped deletes. + PruneDrainRunDelay = 60 * time.Second + // PruneIdleRunDelay paces the runs that delete nothing. Once the backlog is + // gone every run is one of those -- the sweep still visits every stream on its + // cycle -- and a full delay would spend 100 minutes of wall clock a firing + // moving a cursor. Same value as the inter-run delay the trims use. + PruneIdleRunDelay = 5 * time.Second + PruneDrainMaxConsecutiveFailures = 5 ) diff --git a/extensions/tn_digest/scheduler/drain_slot_test.go b/extensions/tn_digest/scheduler/drain_slot_test.go new file mode 100644 index 000000000..f7ccaae65 --- /dev/null +++ b/extensions/tn_digest/scheduler/drain_slot_test.go @@ -0,0 +1,163 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/trufnetwork/kwil-db/common" + "github.com/trufnetwork/kwil-db/config" + "github.com/trufnetwork/kwil-db/core/crypto" + "github.com/trufnetwork/kwil-db/core/crypto/auth" + "github.com/trufnetwork/kwil-db/core/log" + ktypes "github.com/trufnetwork/kwil-db/core/types" + "github.com/trufnetwork/node/extensions/tn_digest/internal" +) + +// The digest drain and the duplicate prune drain share one slot. Both ship with +// the same six-hourly default, so on most firings they want to run at the same +// instant; without the slot they would fetch the same nonce and one would lose. + +func newSlotScheduler() *DigestScheduler { + return NewDigestScheduler(NewDigestSchedulerParams{Logger: log.New(log.WithLevel(log.LevelError))}) +} + +func TestDrainSlot_SecondWaiterBlocksUntilTheFirstReleases(t *testing.T) { + s := newSlotScheduler() + ctx := context.Background() + + if !s.acquireDrainSlot(ctx) { + t.Fatal("first acquire should succeed") + } + + got := make(chan bool, 1) + go func() { got <- s.acquireDrainSlot(ctx) }() + + select { + case <-got: + t.Fatal("second acquire returned while the first drain still held the slot") + case <-time.After(50 * time.Millisecond): + } + + s.releaseDrainSlot() + + select { + case ok := <-got: + if !ok { + t.Fatal("second acquire should have succeeded once the slot was released") + } + case <-time.After(time.Second): + t.Fatal("second acquire never returned after the slot was released") + } + s.releaseDrainSlot() +} + +// A drain waiting behind the other one still has to give up when the node loses +// leadership. Waiting is the right behaviour, waiting forever is not. +func TestDrainSlot_WaiterGivesUpWhenItsContextIsDone(t *testing.T) { + s := newSlotScheduler() + if !s.acquireDrainSlot(context.Background()) { + t.Fatal("first acquire should succeed") + } + defer s.releaseDrainSlot() + + ctx, cancel := context.WithCancel(context.Background()) + got := make(chan bool, 1) + go func() { got <- s.acquireDrainSlot(ctx) }() + cancel() + + select { + case ok := <-got: + if ok { + t.Fatal("a canceled waiter should not report that it took the slot") + } + case <-time.After(time.Second): + t.Fatal("canceled waiter never returned") + } +} + +// Releasing a slot nobody holds is what a drain that gave up on its context does +// on the way out, so it has to be a no-op rather than a block. +func TestDrainSlot_ReleaseWithoutHoldingIsANoOp(t *testing.T) { + s := newSlotScheduler() + s.releaseDrainSlot() + + if !s.acquireDrainSlot(context.Background()) { + t.Fatal("the slot should still be free after a spurious release") + } + s.releaseDrainSlot() +} + +// A scheduler built as a struct literal rather than through the constructor has a +// nil slot. Selecting on a nil channel blocks forever, so the acquire has to make +// one rather than trust the constructor. +func TestDrainSlot_InitialisesWhenTheSchedulerWasBuiltByHand(t *testing.T) { + s := &DigestScheduler{logger: log.New(log.WithLevel(log.LevelError))} + + done := make(chan bool, 1) + go func() { done <- s.acquireDrainSlot(context.Background()) }() + + select { + case ok := <-done: + if !ok { + t.Fatal("acquire on a hand-built scheduler should succeed") + } + case <-time.After(time.Second): + t.Fatal("acquire on a hand-built scheduler blocked") + } + s.releaseDrainSlot() +} + +// gocron's Stop waits for a running job to return, and a sweep waiting on the slot +// returns only when its context is done. So the cancel has to come first, and +// neither call may hold the mutex the job takes on entry. Getting that order wrong +// hangs the node's leadership transition for as long as the other drain runs, +// which is up to a hundred minutes. +// +// The sweep has to be a real cron job for this to reproduce: it is gocron's own +// wait on its running jobs that turns the wrong order into a deadlock. +func TestStopPrune_ReleasesASweepWaitingForTheSlot(t *testing.T) { + s := NewDigestScheduler(NewDigestSchedulerParams{ + Logger: log.New(log.WithLevel(log.LevelError)), + Service: &common.Service{GenesisConfig: &config.GenesisConfig{ChainID: "test-chain"}}, + EngineOps: internal.NewEngineOperations(nil, nil, nil, nil, log.New(log.WithLevel(log.LevelError))), + Signer: stubSigner{}, + Tx: stubBroadcaster{}, + }) + + // The digest drain holds the slot, so the sweep will block on it. + if !s.acquireDrainSlot(context.Background()) { + t.Fatal("could not take the slot for the digest drain") + } + defer s.releaseDrainSlot() + + // Every second, so the job is running well before the stop. + if err := s.StartPrune(context.Background(), "* * * * * *"); err != nil { + t.Fatalf("StartPrune: %v", err) + } + time.Sleep(1500 * time.Millisecond) + + stopped := make(chan error, 1) + go func() { stopped <- s.StopPrune() }() + + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("StopPrune blocked behind a sweep that was waiting for the drain slot") + } +} + +type stubSigner struct{} + +func (stubSigner) Sign(msg []byte) (*auth.Signature, error) { + return &auth.Signature{Data: []byte("sig"), Type: "stub"}, nil +} +func (stubSigner) CompactID() []byte { return []byte("node") } +func (stubSigner) PubKey() crypto.PublicKey { return nil } +func (stubSigner) AuthType() string { return "stub" } + +type stubBroadcaster struct{} + +func (stubBroadcaster) BroadcastTx(ctx context.Context, tx *ktypes.Transaction, sync uint8) (ktypes.Hash, *ktypes.TxResult, error) { + return ktypes.Hash{}, &ktypes.TxResult{Code: uint32(ktypes.CodeOk)}, nil +} diff --git a/extensions/tn_digest/scheduler/scheduler.go b/extensions/tn_digest/scheduler/scheduler.go index 1d9639b4a..c7506dbda 100644 --- a/extensions/tn_digest/scheduler/scheduler.go +++ b/extensions/tn_digest/scheduler/scheduler.go @@ -29,6 +29,21 @@ type DigestScheduler struct { cancel context.CancelFunc mu sync.Mutex + // The duplicate prune sweep runs on its own cron and its own context, because + // duplicate_prune_config carries its own enabled flag and its own schedule. A + // digest config change stops and restarts the digest cron; sharing one would + // make that cancel a prune drain halfway through, and the other way round. + pruneCron *gocron.Scheduler + pruneCtx context.Context + pruneCancel context.CancelFunc + + // drainSlot holds one token and serialises the two drains. They broadcast from + // the same signer account, so two in flight would take the same nonce and one + // would lose; and both delete from primitive_events, so keeping them apart also + // keeps a block from carrying two capped deletes. Both default schedules are + // six-hourly, so without this they would contend on every firing. + drainSlot chan struct{} + broadcaster txBroadcaster signer auth.Signer } @@ -47,6 +62,8 @@ func NewDigestScheduler(params NewDigestSchedulerParams) *DigestScheduler { logger: params.Logger.New("scheduler"), engineOps: params.EngineOps, cron: gocron.NewScheduler(time.UTC), + pruneCron: gocron.NewScheduler(time.UTC), + drainSlot: make(chan struct{}, 1), broadcaster: params.Tx, signer: params.Signer, } @@ -97,6 +114,13 @@ func (s *DigestScheduler) Start(ctx context.Context, cronExpr string) error { } chainID := kwilService.GenesisConfig.ChainID + // One drain at a time; see drainSlot. + if !s.acquireDrainSlot(jobCtx) { + s.logger.Info("digest drain canceled while waiting for the duplicate prune drain") + return + } + defer s.releaseDrainSlot() + // Implement drain mode: run auto_digest repeatedly until has_more=false s.logger.Info("starting digest drain mode", "delete_cap", DigestDeleteCap, @@ -211,17 +235,262 @@ func (s *DigestScheduler) Start(ctx context.Context, cronExpr string) error { return nil } +// Stop stops the digest cron and cancels its drain. It deliberately leaves the +// duplicate prune cron running: the two configurations are independent, and the +// extension restarts the digest cron whenever digest_config changes. +// +// The cancel comes first and neither call happens under the mutex, because +// gocron's Stop waits for a running job to return and this job only returns when +// its context is done. Cancelling afterwards would wait forever, and holding the +// mutex across the wait would block the job in the snapshot it takes on entry. func (s *DigestScheduler) Stop() error { s.mu.Lock() - defer s.mu.Unlock() - s.cron.Stop() - if s.cancel != nil { - s.cancel() + cancel := s.cancel + cron := s.cron + s.mu.Unlock() + + if cancel != nil { + cancel() + } + if cron != nil { + cron.Stop() } s.logger.Info("digest scheduler stopped") return nil } +// acquireDrainSlot blocks until the other drain finishes or ctx is done, and +// reports whether it got the slot. Waiting rather than skipping is deliberate: +// digest and prune ship with the same six-hourly default, so a firing that +// skipped on contention would skip every time. +func (s *DigestScheduler) acquireDrainSlot(ctx context.Context) bool { + s.mu.Lock() + if s.drainSlot == nil { + s.drainSlot = make(chan struct{}, 1) + } + slot := s.drainSlot + s.mu.Unlock() + + select { + case slot <- struct{}{}: + return true + case <-ctx.Done(): + return false + } +} + +func (s *DigestScheduler) releaseDrainSlot() { + s.mu.Lock() + slot := s.drainSlot + s.mu.Unlock() + if slot == nil { + return + } + select { + case <-slot: + default: + } +} + +// StartPrune registers the duplicate prune sweep on its own cron expression. +// +// The extension calls this only when duplicate_prune_config.enabled is true, and +// that column ships false. Nothing on a network prunes until an operator sets it +// through a signed exec-sql. +func (s *DigestScheduler) StartPrune(ctx context.Context, cronExpr string) error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.pruneCancel != nil { + s.pruneCancel() + } + s.pruneCtx, s.pruneCancel = context.WithCancel(ctx) + + if s.pruneCron == nil { + s.pruneCron = gocron.NewScheduler(time.UTC) + } + s.pruneCron.Clear() + + jobCtx := s.pruneCtx + jobFunc := func() { + defer func() { + if r := recover(); r != nil { + s.logger.Error("panic in duplicate prune job", "panic", r, "stack", string(debug.Stack())) + } + }() + s.runPruneDrain(jobCtx) + } + + if j, err := s.pruneCron.Cron(cronExpr).Do(jobFunc); err != nil { + // Fallback for schedules that include seconds. + if j2, err2 := s.pruneCron.CronWithSeconds(cronExpr).Do(jobFunc); err2 != nil { + return fmt.Errorf("register duplicate prune job: %w", err) + } else { + j2.SingletonMode() + } + } else { + j.SingletonMode() + } + + s.pruneCron.StartAsync() + s.logger.Info("duplicate prune scheduler started", "schedule", cronExpr) + return nil +} + +// Running reports whether the digest cron is scheduled, and PruneRunning does the +// same for the duplicate prune sweep. The two crons are independent, so an +// operator or a test that wants to know one is up cannot infer it from the other. +func (s *DigestScheduler) Running() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.cron != nil && s.cron.IsRunning() +} + +func (s *DigestScheduler) PruneRunning() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.pruneCron != nil && s.pruneCron.IsRunning() +} + +// StopPrune stops the duplicate prune cron and cancels a drain in flight. Same +// ordering as Stop, and here it matters more: a sweep can sit for minutes waiting +// on the drain slot, and only its context releases it. +func (s *DigestScheduler) StopPrune() error { + s.mu.Lock() + cancel := s.pruneCancel + cron := s.pruneCron + s.mu.Unlock() + + if cancel != nil { + cancel() + } + if cron != nil { + cron.Stop() + } + s.logger.Info("duplicate prune scheduler stopped") + return nil +} + +// runPruneDrain broadcasts auto_prune_duplicates until the sweep finishes a pass +// over every primitive stream, the run budget is spent, or the context is done. +// +// Unlike digest, finishing early is the exception rather than the rule. The sweep +// is cyclic and has_more_to_delete reports "the cursor has not reached the end of +// a pass", so on a network with more streams than one firing can visit the loop +// runs to PruneDrainMaxRuns every time. That is why an empty run gets the short +// delay: after the backlog is gone every run is an empty one. +func (s *DigestScheduler) runPruneDrain(ctx context.Context) { + s.mu.Lock() + engineOps := s.engineOps + broadcaster := s.broadcaster + signer := s.signer + kwilService := s.kwilService + s.mu.Unlock() + + if engineOps == nil || broadcaster == nil || signer == nil || kwilService == nil || kwilService.GenesisConfig == nil { + s.logger.Warn("duplicate prune job prerequisites missing; skipping run") + return + } + chainID := kwilService.GenesisConfig.ChainID + + // One drain at a time; see drainSlot. + if !s.acquireDrainSlot(ctx) { + s.logger.Info("duplicate prune canceled while waiting for the digest drain") + return + } + defer s.releaseDrainSlot() + + s.logger.Info("starting duplicate prune drain", + "delete_cap", PruneDeleteCap, + "stream_batch_size", PruneStreamBatchSize, + "max_runs", PruneDrainMaxRuns) + + runs := 0 + consecutiveFailures := 0 + totalSweptStreams := 0 + totalEventTimes := 0 + totalRows := 0 + + for runs < PruneDrainMaxRuns { + select { + case <-ctx.Done(): + s.logger.Info("duplicate prune drain canceled", "runs_completed", runs) + return + default: + } + + runs++ + + result, err := engineOps.BroadcastAutoPruneDuplicatesWithRetry( + ctx, + chainID, + signer, + broadcaster.BroadcastTx, + PruneDeleteCap, + PruneStreamBatchSize, + 3, // maxRetries = 3 attempts per run + ) + + delay := PruneIdleRunDelay + if err != nil { + consecutiveFailures++ + s.logger.Warn("auto_prune_duplicates broadcast failed after retries", + "run", runs, + "consecutive_failures", consecutiveFailures, + "error", err) + + if consecutiveFailures >= PruneDrainMaxConsecutiveFailures { + s.logger.Error("too many consecutive failures, aborting duplicate prune drain", + "consecutive_failures", consecutiveFailures, + "max_allowed", PruneDrainMaxConsecutiveFailures) + return + } + delay = PruneDrainRunDelay + } else { + consecutiveFailures = 0 + totalSweptStreams += result.SweptStreams + totalEventTimes += result.DeletedEventTimes + totalRows += result.DeletedRows + + s.logger.Info("duplicate prune run completed", + "run", runs, + "swept_streams", result.SweptStreams, + "deleted_event_times", result.DeletedEventTimes, + "deleted_rows", result.DeletedRows, + "has_more", result.HasMoreToDelete, + "cumulative_swept", totalSweptStreams, + "cumulative_deleted_rows", totalRows) + + if !result.HasMoreToDelete { + s.logger.Info("duplicate prune pass completed", + "total_runs", runs, + "total_swept_streams", totalSweptStreams, + "total_deleted_event_times", totalEventTimes, + "total_deleted_rows", totalRows) + return + } + + if result.DeletedRows > 0 { + delay = PruneDrainRunDelay + } + } + + select { + case <-ctx.Done(): + s.logger.Info("duplicate prune drain canceled during sleep", "runs_completed", runs) + return + case <-time.After(delay): + } + } + + s.logger.Info("duplicate prune drain reached max runs", + "max_runs", PruneDrainMaxRuns, + "runs_completed", runs, + "total_swept_streams", totalSweptStreams, + "total_deleted_event_times", totalEventTimes, + "total_deleted_rows", totalRows) +} + // trimOrderEvents runs the trim_order_events action in a drain loop (best-effort). // Called after digest drain completes. Failures are logged but do not fail the digest job. func (s *DigestScheduler) trimOrderEvents( @@ -339,3 +608,17 @@ func (s *DigestScheduler) RunOnce(ctx context.Context) error { chainID := s.kwilService.GenesisConfig.ChainID return s.engineOps.BuildAndBroadcastAutoDigestTx(ctx, chainID, s.signer, s.broadcaster.BroadcastTx) } + +// RunPruneOnce broadcasts a single auto_prune_duplicates batch (for tests and +// manual triggering). It takes no drain slot: a caller reaching for one batch is +// not the scheduler, and blocking it behind a six-hour drain would be surprising. +func (s *DigestScheduler) RunPruneOnce(ctx context.Context) (*internal.PruneTxResult, error) { + if s.engineOps == nil || s.broadcaster == nil || s.signer == nil || s.kwilService == nil || s.kwilService.GenesisConfig == nil { + return nil, fmt.Errorf("missing prerequisites to run duplicate prune once") + } + chainID := s.kwilService.GenesisConfig.ChainID + return s.engineOps.BroadcastAutoPruneDuplicatesWithRetry( + ctx, chainID, s.signer, s.broadcaster.BroadcastTx, + PruneDeleteCap, PruneStreamBatchSize, 3, + ) +} diff --git a/extensions/tn_digest/scheduler_lifecycle.go b/extensions/tn_digest/scheduler_lifecycle.go index e59e833a2..0a5d252b9 100644 --- a/extensions/tn_digest/scheduler_lifecycle.go +++ b/extensions/tn_digest/scheduler_lifecycle.go @@ -50,6 +50,16 @@ func (e *Extension) stopSchedulerIfRunning() { } } +func (e *Extension) startPruneScheduler(_ context.Context) error { + return e.Scheduler().StartPrune(context.Background(), e.PruneSchedule()) +} + +func (e *Extension) stopPruneIfRunning() { + if e.Scheduler() != nil { + _ = e.Scheduler().StopPrune() + } +} + // wireSignerAndBroadcaster fills in signer and broadcaster if not already set. func wireSignerAndBroadcaster(app *common.App, ext *Extension) { if app == nil || app.Service == nil || app.Service.LocalConfig == nil { diff --git a/extensions/tn_digest/tn_digest.go b/extensions/tn_digest/tn_digest.go index eb32564c9..6cf92d744 100644 --- a/extensions/tn_digest/tn_digest.go +++ b/extensions/tn_digest/tn_digest.go @@ -70,12 +70,20 @@ func engineReadyHook(ctx context.Context, app *common.App) error { schedule = DefaultDigestSchedule } + // The duplicate prune sweep has its own table, its own enabled flag and its own + // schedule, so it is snapshotted separately rather than derived from digest's. + pruneEnabled, pruneSchedule, _ := engOps.LoadPruneConfig(ctx) + if pruneSchedule == "" { + pruneSchedule = DefaultPruneSchedule + } + // Create extension instance and snapshot references ext := GetExtension() ext.logger = logger ext.SetService(app.Service) ext.SetEngineOps(engOps) ext.SetConfig(enabled, schedule) + ext.SetPruneConfig(pruneEnabled, pruneSchedule) // Load config from node TOML [extensions.tn_digest] if ext.Service() != nil && ext.Service().LocalConfig != nil { @@ -135,7 +143,9 @@ func digestLeaderAcquire(ctx context.Context, app *common.App, block *common.Blo return } ext.setLeader(true) - if !ext.ConfigEnabled() { + // Either feature being on is reason enough to build the scheduler; both off + // leaves it nil, so a node with nothing enabled allocates nothing. + if !ext.ConfigEnabled() && !ext.PruneEnabled() { return } service := ext.Service() @@ -152,10 +162,19 @@ func digestLeaderAcquire(ctx context.Context, app *common.App, block *common.Blo ext.Logger().Debug("tn_digest: prerequisites missing; deferring start until broadcaster/signer/engine/service are available") return } - if err := ext.startScheduler(ctx); err != nil { - ext.Logger().Warn("failed to start tn_digest scheduler on leader acquire", "error", err) - } else { - ext.Logger().Info("tn_digest started (leader)", "schedule", ext.Schedule()) + if ext.ConfigEnabled() { + if err := ext.startScheduler(ctx); err != nil { + ext.Logger().Warn("failed to start tn_digest scheduler on leader acquire", "error", err) + } else { + ext.Logger().Info("tn_digest started (leader)", "schedule", ext.Schedule()) + } + } + if ext.PruneEnabled() { + if err := ext.startPruneScheduler(ctx); err != nil { + ext.Logger().Warn("failed to start duplicate prune scheduler on leader acquire", "error", err) + } else { + ext.Logger().Info("duplicate prune started (leader)", "schedule", ext.PruneSchedule()) + } } } @@ -165,6 +184,7 @@ func digestLeaderLose(ctx context.Context, app *common.App, block *common.BlockC return } ext.setLeader(false) + ext.stopPruneIfRunning() ext.stopSchedulerIfRunning() if ext.Logger() != nil { ext.Logger().Info("tn_digest stopped (lost leadership)") @@ -206,8 +226,20 @@ func digestLeaderEndBlock(ctx context.Context, app *common.App, block *common.Bl return } + pruneEnabled, pruneSchedule, pruneLoadErr := ext.EngineOps().LoadPruneConfig(ctx) + if pruneLoadErr != nil { + // The digest config did load, so apply it rather than dropping it, and let + // the retry worker come back for both. + ext.applyConfigChangeWithLock(ctx, enabled, schedule, app) + ext.Logger().Warn("duplicate prune config reload failed in end-block, signaling background retry worker", "error", pruneLoadErr) + ext.SetLastCheckedHeight(block.Height) + ext.signalRetryNeeded() + return + } + // Apply config change with proper synchronization (prevents race with background worker) ext.applyConfigChangeWithLock(ctx, enabled, schedule, app) + ext.applyPruneConfigChangeWithLock(ctx, pruneEnabled, pruneSchedule, app) ext.SetLastCheckedHeight(block.Height) }