feat: run a duplicate prune sweep from the node

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)

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
 )

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