diff --git a/extensions/tn_digest/README.md b/extensions/tn_digest/README.md index cbec5a79..48eae46a 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 16ab7fb6..4520cee1 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 5736fec0..9a9c7177 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/internal/engine_ops.go b/extensions/tn_digest/internal/engine_ops.go index 64b15762..ffb748da 100644 --- a/extensions/tn_digest/internal/engine_ops.go +++ b/extensions/tn_digest/internal/engine_ops.go @@ -808,3 +808,238 @@ func parseDigestResultFromTxLog(logOutput string) (*DigestTxResult, error) { return result, nil } + +// PruneTxResult represents the parsed result from an auto_prune_duplicates +// transaction. +type PruneTxResult struct { + SweptStreams int + DeletedEventTimes int + DeletedRows int + HasMoreToDelete bool +} + +// LoadPruneConfig reads the single-row duplicate prune configuration. +// Returns (enabled, schedule). If the table or row is missing it returns +// false, "" and no error, so a node running a binary newer than its migrations +// simply leaves the sweep off. +func (e *EngineOperations) LoadPruneConfig(ctx context.Context) (bool, string, error) { + var ( + enabled bool + schedule string + found bool + ) + + db, cleanup, err := e.getFreshReadTx(ctx) + if err != nil { + return false, "", fmt.Errorf("get fresh read tx: %w", err) + } + defer cleanup() + + err = e.engine.ExecuteWithoutEngineCtx(ctx, db, + `SELECT enabled, prune_schedule FROM main.duplicate_prune_config WHERE id = 1`, nil, + func(row *common.Row) error { + if len(row.Values) >= 2 { + if v, ok := row.Values[0].(bool); ok { + enabled = v + } + if s, ok := row.Values[1].(string); ok { + schedule = s + } + found = true + } + return nil + }) + if err != nil { + msg := err.Error() + // Tolerate a missing table the way LoadDigestConfig does; everything else + // should surface to the caller. + if strings.Contains(msg, "duplicate_prune_config") && (strings.Contains(msg, "does not exist") || strings.Contains(msg, "undefined") || strings.Contains(msg, "not found")) { + e.logger.Info("duplicate_prune_config table not found; duplicate pruning stays off") + return false, "", nil + } + return false, "", err + } + if !found { + return false, "", nil + } + return enabled, schedule, nil +} + +// BroadcastAutoPruneDuplicatesWithRetry wraps broadcastAutoPruneDuplicatesOnce +// with retry logic. On any error it re-fetches a fresh nonce before retrying, +// mirroring BroadcastTrimOrderEventsWithRetry. +func (e *EngineOperations) BroadcastAutoPruneDuplicatesWithRetry( + ctx context.Context, + chainID string, + signer auth.Signer, + broadcaster func(context.Context, *ktypes.Transaction, uint8) (ktypes.Hash, *ktypes.TxResult, error), + deleteCap int, + streamBatchSize int, + maxRetries int, +) (*PruneTxResult, error) { + var lastErr error + backoff := 5 * time.Second + maxBackoff := 30 * time.Second + + for attempt := 0; attempt <= maxRetries; attempt++ { + if attempt > 0 { + e.logger.Warn("retrying auto_prune_duplicates with fresh nonce", + "attempt", attempt, "last_error", lastErr) + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(backoff): + } + + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } + + result, err := e.broadcastAutoPruneDuplicatesOnce(ctx, chainID, signer, broadcaster, deleteCap, streamBatchSize) + if err == nil { + return result, nil + } + lastErr = err + } + + return nil, fmt.Errorf("auto_prune_duplicates max retries (%d) exceeded: %w", maxRetries, lastErr) +} + +// broadcastAutoPruneDuplicatesOnce fetches a fresh nonce and broadcasts once. +// +// The action's third parameter, retention_days, is deliberately left off the +// argument list so its NULL default applies and the action reads retention from +// duplicate_prune_config. Retention is an operator setting that has to be +// changeable without a binary release, and passing it from here would silently +// override whatever the operator set. +func (e *EngineOperations) broadcastAutoPruneDuplicatesOnce( + ctx context.Context, + chainID string, + signer auth.Signer, + broadcaster func(context.Context, *ktypes.Transaction, uint8) (ktypes.Hash, *ktypes.TxResult, error), + deleteCap int, + streamBatchSize int, +) (*PruneTxResult, error) { + signerAccountID, err := ktypes.GetSignerAccount(signer) + if err != nil { + return nil, fmt.Errorf("get signer account: %w", err) + } + + db, cleanup, err := e.getFreshReadTx(ctx) + if err != nil { + return nil, fmt.Errorf("get fresh read tx: %w", err) + } + defer cleanup() + + account, err := e.accounts.GetAccount(ctx, db, signerAccountID) + var nextNonce uint64 + if err != nil { + msg := strings.ToLower(err.Error()) + if !strings.Contains(msg, "not found") && !strings.Contains(msg, "no rows") { + return nil, fmt.Errorf("get account: %w", err) + } + nextNonce = 1 + } else { + nextNonce = uint64(account.Nonce + 1) + } + + deleteCapArg, err := ktypes.EncodeValue(int64(deleteCap)) + if err != nil { + return nil, fmt.Errorf("encode deleteCap: %w", err) + } + streamBatchSizeArg, err := ktypes.EncodeValue(int64(streamBatchSize)) + if err != nil { + return nil, fmt.Errorf("encode streamBatchSize: %w", err) + } + + payload := &ktypes.ActionExecution{ + Namespace: "main", + Action: "auto_prune_duplicates", + Arguments: [][]*ktypes.EncodedValue{{ + deleteCapArg, streamBatchSizeArg, + }}, + } + + tx, err := ktypes.CreateNodeTransaction(payload, chainID, nextNonce) + if err != nil { + return nil, fmt.Errorf("create tx: %w", err) + } + if err := tx.Sign(signer); err != nil { + return nil, fmt.Errorf("sign tx: %w", err) + } + + hash, txResult, err := broadcaster(ctx, tx, 1) + if err != nil { + return nil, fmt.Errorf("broadcast tx: %w", err) + } + + if txResult.Code != uint32(ktypes.CodeOk) { + return nil, fmt.Errorf("transaction failed with code %d: %s", + txResult.Code, txResult.Log) + } + + result, err := parsePruneResultFromTxLog(txResult.Log) + if err != nil { + e.logger.Warn("failed to parse auto_prune_duplicates result", "error", err, "log", txResult.Log) + return nil, fmt.Errorf("parse prune result: %w", err) + } + + e.logger.Info("auto_prune_duplicates completed", + "swept_streams", result.SweptStreams, + "deleted_event_times", result.DeletedEventTimes, + "deleted_rows", result.DeletedRows, + "has_more", result.HasMoreToDelete, + "tx_hash", hash.String(), + "nonce", nextNonce, + "delete_cap", deleteCap, + "stream_batch_size", streamBatchSize) + + return result, nil +} + +// parsePruneResultFromTxLog extracts the sweep counters from the NOTICE the +// action emits. The payload is JSON rather than the key=value the trim actions +// use, so this mirrors parseDigestResultFromTxLog rather than +// parseTrimResultFromTxLog. +func parsePruneResultFromTxLog(logOutput string) (*PruneTxResult, error) { + if logOutput == "" { + return nil, fmt.Errorf("empty log output") + } + + var pruneJSON string + for _, line := range strings.Split(logOutput, "\n") { + if !strings.Contains(line, "auto_prune_duplicates:") { + continue + } + parts := strings.SplitN(line, "auto_prune_duplicates:", 2) + if len(parts) == 2 { + pruneJSON = strings.TrimSpace(parts[1]) + } + } + + if pruneJSON == "" { + return nil, fmt.Errorf("no auto_prune_duplicates log entry found in: %q", logOutput) + } + + pruneJSON = strings.Trim(pruneJSON, `"`) + + var jsonResult struct { + SweptStreams int `json:"swept_streams"` + DeletedEventTimes int `json:"deleted_event_times"` + DeletedRows int `json:"deleted_rows"` + HasMoreToDelete bool `json:"has_more_to_delete"` + } + if err := json.Unmarshal([]byte(pruneJSON), &jsonResult); err != nil { + return nil, fmt.Errorf("failed to parse prune JSON: %w", err) + } + + return &PruneTxResult{ + SweptStreams: jsonResult.SweptStreams, + DeletedEventTimes: jsonResult.DeletedEventTimes, + DeletedRows: jsonResult.DeletedRows, + HasMoreToDelete: jsonResult.HasMoreToDelete, + }, nil +} diff --git a/extensions/tn_digest/internal/prune_ops_test.go b/extensions/tn_digest/internal/prune_ops_test.go new file mode 100644 index 00000000..6eff78f7 --- /dev/null +++ b/extensions/tn_digest/internal/prune_ops_test.go @@ -0,0 +1,193 @@ +package internal + +import ( + "context" + "errors" + "testing" + "time" + + "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" +) + +// An action's return value is not visible to a transaction's broadcaster, so the +// sweep's counters reach the scheduler only through the NOTICE the action emits. +// Parsing it is the whole interface between the two. + +const pruneNotice = `auto_prune_duplicates:{"swept_streams":100,"deleted_event_times":42,"deleted_rows":57,"has_more_to_delete":true}` + +func TestParsePruneResultFromTxLog_ReadsTheSweepCounters(t *testing.T) { + res, err := parsePruneResultFromTxLog("INFO something\nNOTICE: " + pruneNotice + "\nother") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if res.SweptStreams != 100 { + t.Fatalf("swept_streams: want 100, got %d", res.SweptStreams) + } + if res.DeletedEventTimes != 42 { + t.Fatalf("deleted_event_times: want 42, got %d", res.DeletedEventTimes) + } + if res.DeletedRows != 57 { + t.Fatalf("deleted_rows: want 57, got %d", res.DeletedRows) + } + if !res.HasMoreToDelete { + t.Fatalf("has_more_to_delete: want true, got false") + } +} + +// has_more_to_delete false is what ends a drain, so reading it wrong would either +// spin the loop to its run budget every firing or stop a sweep on its first batch. +func TestParsePruneResultFromTxLog_ReadsTheEndOfAPass(t *testing.T) { + res, err := parsePruneResultFromTxLog(`auto_prune_duplicates:{"swept_streams":7,"deleted_event_times":0,"deleted_rows":0,"has_more_to_delete":false}`) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if res.HasMoreToDelete { + t.Fatalf("has_more_to_delete: want false, got true") + } + if res.SweptStreams != 7 { + t.Fatalf("swept_streams: want 7, got %d", res.SweptStreams) + } +} + +// A transaction that committed without the notice means the action did not run the +// way this code assumes. Treating that as a zero-valued success would report a +// finished pass and silently stop the drain. +func TestParsePruneResultFromTxLog_NoEntry(t *testing.T) { + if _, err := parsePruneResultFromTxLog("INFO: nothing relevant here\nNOTICE: auto_digest:{}"); err == nil { + t.Fatal("expected an error for a log with no auto_prune_duplicates entry") + } +} + +// The digest marker is not a prefix of this one, but both parsers scan the same +// log, so it is worth pinning that neither reads the other's line. +func TestParsePruneResultFromTxLog_IgnoresTheDigestNotice(t *testing.T) { + log := "NOTICE: auto_digest:{\"processed_days\":2,\"total_deleted_rows\":500,\"has_more_to_delete\":false}\nNOTICE: " + pruneNotice + res, err := parsePruneResultFromTxLog(log) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if res.DeletedRows != 57 { + t.Fatalf("read the digest line: deleted_rows want 57, got %d", res.DeletedRows) + } + + digestRes, err := parseDigestResultFromTxLog(log) + if err != nil { + t.Fatalf("unexpected error reading the digest notice: %v", err) + } + if digestRes.TotalDeletedRows != 500 { + t.Fatalf("read the prune line: total_deleted_rows want 500, got %d", digestRes.TotalDeletedRows) + } +} + +type prunePathBroadcaster struct { + attempts int + failUntil int + // action and argCount record what the last transaction actually asked for. + action string + argCount int +} + +func (m *prunePathBroadcaster) broadcast(ctx context.Context, tx *ktypes.Transaction, sync uint8) (ktypes.Hash, *ktypes.TxResult, error) { + m.attempts++ + + if payload := new(ktypes.ActionExecution); payload.UnmarshalBinary(tx.Body.Payload) == nil { + m.action = payload.Action + if len(payload.Arguments) == 1 { + m.argCount = len(payload.Arguments[0]) + } + } + + result := &ktypes.TxResult{Code: uint32(ktypes.CodeOk), Log: pruneNotice} + if m.attempts <= m.failUntil { + return ktypes.Hash{}, result, errors.New("network error") + } + return ktypes.Hash{1, 2, 3}, result, nil +} + +func newPruneSigner(t *testing.T) auth.Signer { + t.Helper() + priv, _, err := crypto.GenerateSecp256k1Key(nil) + if err != nil { + t.Fatalf("generate key: %v", err) + } + return auth.GetNodeSigner(priv) +} + +// retention_days is deliberately not passed. It is the third parameter and it +// defaults to NULL, which makes the action read retention from +// duplicate_prune_config -- so an operator can change it with a signed exec-sql +// instead of a binary release. Sending two arguments is what keeps that true. +func TestBroadcastAutoPruneDuplicates_LeavesRetentionToTheConfig(t *testing.T) { + accounts := &mockAccounts{} + broadcaster := &prunePathBroadcaster{} + ops := &EngineOperations{logger: log.New(), accounts: accounts} + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + result, err := ops.BroadcastAutoPruneDuplicatesWithRetry( + ctx, "test-chain", newPruneSigner(t), broadcaster.broadcast, 100000, 100, 3, + ) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + if result.DeletedRows != 57 { + t.Fatalf("deleted_rows: want 57, got %d", result.DeletedRows) + } + if broadcaster.action != "auto_prune_duplicates" { + t.Fatalf("action: want auto_prune_duplicates, got %q", broadcaster.action) + } + if broadcaster.argCount != 2 { + t.Fatalf("argument count: want 2 so retention_days keeps its NULL default, got %d", broadcaster.argCount) + } + if broadcaster.attempts != 1 { + t.Fatalf("attempts: want 1, got %d", broadcaster.attempts) + } +} + +// Each retry refetches the nonce rather than reusing the one that just lost, which +// is what makes a collision with the digest drain recoverable rather than fatal. +func TestBroadcastAutoPruneDuplicates_RefetchesTheNonceOnRetry(t *testing.T) { + accounts := &mockAccounts{} + // One failure, not two: the claim is that a retry refetches, and the backoff + // before the second attempt is a real five seconds of test time. + broadcaster := &prunePathBroadcaster{failUntil: 1} + ops := &EngineOperations{logger: log.New(), accounts: accounts} + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + if _, err := ops.BroadcastAutoPruneDuplicatesWithRetry( + ctx, "test-chain", newPruneSigner(t), broadcaster.broadcast, 100000, 100, 3, + ); err != nil { + t.Fatalf("expected success after a retry, got %v", err) + } + if broadcaster.attempts != 2 { + t.Fatalf("attempts: want 2, got %d", broadcaster.attempts) + } + if accounts.nonceCalls != 2 { + t.Fatalf("nonce fetches: want one per attempt (2), got %d", accounts.nonceCalls) + } +} + +// A drain that outlives its leadership has to stop rather than keep broadcasting. +func TestBroadcastAutoPruneDuplicates_StopsOnContextCancellation(t *testing.T) { + accounts := &mockAccounts{} + broadcaster := &prunePathBroadcaster{failUntil: 100} + ops := &EngineOperations{logger: log.New(), accounts: accounts} + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + if _, err := ops.BroadcastAutoPruneDuplicatesWithRetry( + ctx, "test-chain", newPruneSigner(t), broadcaster.broadcast, 100000, 100, 5, + ); err == nil { + t.Fatal("expected an error once the context was canceled") + } + if broadcaster.attempts > 2 { + t.Fatalf("kept broadcasting past cancellation: %d attempts", broadcaster.attempts) + } +} diff --git a/extensions/tn_digest/leader_reload_test.go b/extensions/tn_digest/leader_reload_test.go index 9c8329ae..aa88dc42 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 00000000..6894e449 --- /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 f258c495..abcc92ea 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 00000000..076b9456 --- /dev/null +++ b/extensions/tn_digest/scheduler/drain_slot_test.go @@ -0,0 +1,244 @@ +package scheduler + +import ( + "context" + "math/big" + "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/kwil-db/node/types/sql" + "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: testSigner(), + 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") + } +} + +// RunPruneOnce broadcasts from the same signer account as the scheduled drains, so +// letting it run alongside one means both read the same nonce and one transaction +// loses. It refuses instead of waiting, and the error says which of the two it was. +func TestRunPruneOnce_RefusesWhileADrainHoldsTheSlot(t *testing.T) { + s := newPruneTestScheduler() + + if !s.acquireDrainSlot(context.Background()) { + t.Fatal("could not take the slot for the drain") + } + + done := make(chan error, 1) + go func() { + _, err := s.RunPruneOnce(context.Background()) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("RunPruneOnce should have refused while a drain held the slot") + } + case <-time.After(2 * time.Second): + t.Fatal("RunPruneOnce waited for the slot instead of refusing") + } + + // And it leaves the slot where it found it, so the drain still owns it. + s.releaseDrainSlot() + if !s.tryAcquireDrainSlot() { + t.Fatal("the refused call consumed or corrupted the slot") + } + s.releaseDrainSlot() +} + +// The other half of the same claim: refusing is about contention, not a permanent +// state, so a free slot lets the one-off through and hands it back afterwards. +func TestRunPruneOnce_RunsAndReleasesWhenTheSlotIsFree(t *testing.T) { + s := newPruneTestScheduler() + + if _, err := s.RunPruneOnce(context.Background()); err != nil { + t.Fatalf("RunPruneOnce with a free slot: %v", err) + } + if !s.tryAcquireDrainSlot() { + t.Fatal("RunPruneOnce did not release the slot") + } + s.releaseDrainSlot() +} + +// newPruneTestScheduler builds a scheduler whose dependencies are present but inert, +// which is enough for the entry points that only broadcast. +func newPruneTestScheduler() *DigestScheduler { + return NewDigestScheduler(NewDigestSchedulerParams{ + Logger: log.New(log.WithLevel(log.LevelError)), + Service: &common.Service{GenesisConfig: &config.GenesisConfig{ChainID: "test-chain"}}, + EngineOps: internal.NewEngineOperations(nil, nil, nil, stubAccounts{}, log.New(log.WithLevel(log.LevelError))), + Signer: testSigner(), + Tx: stubBroadcaster{}, + }) +} + +type stubAccounts struct{} + +func (stubAccounts) GetAccount(ctx context.Context, db sql.Executor, id *ktypes.AccountID) (*ktypes.Account, error) { + return &ktypes.Account{ID: id, Nonce: 0, Balance: big.NewInt(1000)}, nil +} +func (stubAccounts) Credit(ctx context.Context, db sql.Executor, id *ktypes.AccountID, amt *big.Int) error { + return nil +} +func (stubAccounts) Transfer(ctx context.Context, db sql.TxMaker, from, to *ktypes.AccountID, amt *big.Int) error { + return nil +} +func (stubAccounts) ApplySpend(ctx context.Context, db sql.Executor, id *ktypes.AccountID, amt *big.Int, nonce int64) error { + return nil +} + +// testSigner is a real key rather than a stub. A stub with a nil PubKey panics +// inside GetSignerAccount, which the slot tests never reach but the one-off +// entry points do. +func testSigner() auth.Signer { + priv, _, err := crypto.GenerateSecp256k1Key(nil) + if err != nil { + panic(err) + } + return auth.GetNodeSigner(priv) +} + +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), + Log: `auto_prune_duplicates:{"swept_streams":1,"deleted_event_times":0,"deleted_rows":0,"has_more_to_delete":false}`, + }, nil +} diff --git a/extensions/tn_digest/scheduler/scheduler.go b/extensions/tn_digest/scheduler/scheduler.go index 1d9639b4..2983fe12 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,282 @@ 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 } +// slotChan returns the drain slot, creating it if the scheduler was built as a +// struct literal rather than through the constructor. Selecting on a nil channel +// blocks forever, so this cannot trust the constructor to have run. +func (s *DigestScheduler) slotChan() chan struct{} { + s.mu.Lock() + defer s.mu.Unlock() + if s.drainSlot == nil { + s.drainSlot = make(chan struct{}, 1) + } + return s.drainSlot +} + +// 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 { + slot := s.slotChan() + select { + case slot <- struct{}{}: + return true + case <-ctx.Done(): + return false + } +} + +// tryAcquireDrainSlot takes the slot only if it is free. It is for the one-off +// entry points, which need the same protection against two transactions taking +// the same nonce but should not disappear into a drain that can hold the slot for +// the better part of two hours. +func (s *DigestScheduler) tryAcquireDrainSlot() bool { + slot := s.slotChan() + select { + case slot <- struct{}{}: + return true + default: + 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 +628,27 @@ 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 the drain slot, because it broadcasts from the same signer account as +// the scheduled drains and would otherwise read the same nonce. It does not wait +// for it: a caller asking for one batch wants an answer, and a drain holds the +// slot for as long as it runs. Refusing says which of the two happened, where a +// nonce collision would only show up as a retry in the logs. +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") + } + if !s.tryAcquireDrainSlot() { + return nil, fmt.Errorf("a digest or duplicate prune drain is already running; retry once it finishes") + } + defer s.releaseDrainSlot() + + 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 e59e833a..0a5d252b 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 eb32564c..6cf92d74 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) } diff --git a/tests/streams/digest/prune_actions_test.go b/tests/streams/digest/prune_actions_test.go index da4d2044..9d0201ff 100644 --- a/tests/streams/digest/prune_actions_test.go +++ b/tests/streams/digest/prune_actions_test.go @@ -49,6 +49,7 @@ func TestPruneActions(t *testing.T) { SeedStatements: migrations.GetSeedScriptStatements(), FunctionTests: []kwilTesting.TestFunc{ WithPruneStream(testPruneCollapsesRunsAndLeavesReadsAlone(t)), + WithPruneStream(testPruneLeavesEveryReadPathAlone(t)), WithPruneStream(testPruneKeepsTheFirstAndNewestRecords(t)), WithPruneStream(testPruneKeepsTheTruflationWatermark(t)), WithPruneStream(testPruneLeavesRecentRecordsAlone(t)), @@ -141,6 +142,129 @@ func testPruneCollapsesRunsAndLeavesReadsAlone(t *testing.T) func(context.Contex } } +// The claim widened to the reads a consumer actually makes, and driven the way the +// scheduler drives it: auto_prune_duplicates with retention left NULL so it comes +// from duplicate_prune_config, looped until the sweep reports a finished pass. +// +// Values are compared, not whole rows. An anchored read reports the anchor's own +// event_time, and pruning moves that back to the head of the run on purpose, so +// the timestamp beside a value is expected to move. get_first_record is left out +// for a stronger reason: it is a forward scan rather than an anchored lookup, so +// pruning the record it would have returned moves its value, and migration 057's +// header says so. +func testPruneLeavesEveryReadPathAlone(t *testing.T) func(context.Context, *kwilTesting.Platform) error { + return func(ctx context.Context, platform *kwilTesting.Platform) error { + streamRef, err := setup.GetStreamIdForDeployer(ctx, platform, pruneStreamName) + if err != nil { + return errors.Wrap(err, "resolve stream ref") + } + if err := seedPruneRecords(ctx, platform, streamRef, []pruneRecord{ + {Day: 1, Value: "10"}, + {Day: 2, Value: "10"}, + {Day: 3, Value: "10"}, + {Day: 4, Value: "12"}, + {Day: 5, Value: "12"}, + {Day: 6, Value: "12"}, + {Day: 7, Value: "8"}, + {Day: 8, Value: "8"}, + {Day: 9, Value: "11"}, + {Day: 10, Value: "11"}, + {Day: 11, Value: "11"}, + {Day: 12, Value: "11"}, + }); err != nil { + return err + } + + days := []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12} + // Windows worth naming. Days 2 to 3 sit entirely inside a run, so after + // pruning they hold no record of their own and the answer can only come + // from the anchor -- which is the read B1 fixed and the reason pruning was + // safe to build at all. + windows := [][2]int64{{1, 12}, {2, 3}, {5, 8}, {9, 12}} + + beforeRecords, err := readEachDay(ctx, platform, days) + if err != nil { + return errors.Wrap(err, "read records before pruning") + } + beforeIndex, err := readIndexEachDay(ctx, platform, days) + if err != nil { + return errors.Wrap(err, "read index before pruning") + } + beforeExtremes, err := readExtremesOverWindows(ctx, platform, windows) + if err != nil { + return errors.Wrap(err, "read high and low before pruning") + } + + storedBefore, err := readStoredRecords(ctx, platform, streamRef) + if err != nil { + return err + } + + // A cap of two forces the resume path, so this covers the same loop the + // scheduler runs rather than a single call that happens to finish. + rounds := 0 + for { + rounds++ + if rounds > 20 { + return errors.New("the sweep never reported a finished pass") + } + res, _, err := callAutoPrune(ctx, platform, 2, 100, nil) + if err != nil { + return errors.Wrapf(err, "sweep %d", rounds) + } + if len(res) != 1 { + return errors.Errorf("sweep %d returned %d rows, want 1", rounds, len(res)) + } + if res[0][3] == "false" { + break + } + } + if rounds < 2 { + return errors.Errorf("the cap should have taken more than one round, got %d", rounds) + } + + storedAfter, err := readStoredRecords(ctx, platform, streamRef) + if err != nil { + return err + } + if got, want := storedAfter, "86400=10 345600=12 604800=8 777600=11 1036800=11"; got != want { + return errors.Errorf("wrong records survived:\n got: %s\nwant: %s", got, want) + } + if storedBefore == storedAfter { + // Without this the three comparisons below would pass on a sweep that + // deleted nothing, which is the one way this could look green while + // proving nothing. + return errors.New("the sweep deleted nothing, so the comparisons prove nothing") + } + + afterRecords, err := readEachDay(ctx, platform, days) + if err != nil { + return errors.Wrap(err, "read records after pruning") + } + if beforeRecords != afterRecords { + return errors.Errorf("pruning moved get_record:\nbefore: %s\n after: %s", beforeRecords, afterRecords) + } + + afterIndex, err := readIndexEachDay(ctx, platform, days) + if err != nil { + return errors.Wrap(err, "read index after pruning") + } + if beforeIndex != afterIndex { + return errors.Errorf("pruning moved get_index:\nbefore: %s\n after: %s", beforeIndex, afterIndex) + } + + afterExtremes, err := readExtremesOverWindows(ctx, platform, windows) + if err != nil { + return errors.Wrap(err, "read high and low after pruning") + } + if beforeExtremes != afterExtremes { + return errors.Errorf("pruning moved get_high_value or get_low_value:\nbefore: %s\n after: %s", + beforeExtremes, afterExtremes) + } + return nil + } +} + // ============================================================================= // The records that are never candidates // ============================================================================= @@ -1092,6 +1216,59 @@ func readEachDay(ctx context.Context, platform *kwilTesting.Platform, days []int return strings.Join(parts, " "), nil } +// readIndexEachDay is readEachDay through get_index. base_time is left NULL so the +// base resolves the way a consumer's would: from default_base_time if the stream +// has one, otherwise from the first record -- which rule 3 never deletes. +func readIndexEachDay(ctx context.Context, platform *kwilTesting.Platform, days []int64) (string, error) { + address, err := util.NewEthereumAddressFromBytes(platform.Deployer) + if err != nil { + return "", errors.Wrap(err, "deployer address") + } + var parts []string + for _, day := range days { + at := day * daySecs + rows, err := callActionAsStrings(ctx, platform, "get_index", 2, + address.Address(), pruneStreamId.String(), at, at, nil, nil) + if err != nil { + return "", errors.Wrapf(err, "get_index at %d", at) + } + if len(rows) != 1 { + return "", errors.Errorf("get_index answered %d rows for day %d; expected exactly one", len(rows), day) + } + parts = append(parts, fmt.Sprintf("%d->%s", day, rows[0][1])) + } + return strings.Join(parts, " "), nil +} + +// readExtremesOverWindows renders get_high_value and get_low_value across each +// window as "from-to->high/low". Both anchor since B1, so a window holding no +// record of its own still answers with the value carried into it. +func readExtremesOverWindows(ctx context.Context, platform *kwilTesting.Platform, windows [][2]int64) (string, error) { + address, err := util.NewEthereumAddressFromBytes(platform.Deployer) + if err != nil { + return "", errors.Wrap(err, "deployer address") + } + var parts []string + for _, window := range windows { + from, to := window[0]*daySecs, window[1]*daySecs + values := make([]string, 0, 2) + for _, action := range []string{"get_high_value", "get_low_value"} { + rows, err := callActionAsStrings(ctx, platform, action, 2, + address.Address(), pruneStreamId.String(), from, to, nil) + if err != nil { + return "", errors.Wrapf(err, "%s over days %d-%d", action, window[0], window[1]) + } + if len(rows) != 1 { + return "", errors.Errorf("%s answered %d rows for days %d-%d; expected exactly one", + action, len(rows), window[0], window[1]) + } + values = append(values, rows[0][1]) + } + parts = append(parts, fmt.Sprintf("%d-%d->%s/%s", window[0], window[1], values[0], values[1])) + } + return strings.Join(parts, " "), nil +} + func callBatchPrune(ctx context.Context, platform *kwilTesting.Platform, streamRefs []int, retentionSeconds int64, deleteCap int) ([]procedure.ResultRow, error) { return callActionAsStrings(ctx, platform, "batch_prune_duplicates", 3, streamRefs, retentionSeconds, deleteCap) }