From 52211065324a036dc7c2e3acec92c0c7767adc1b Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Tue, 8 Sep 2026 08:25:03 +0700 Subject: [PATCH] chore: size the duplicate prune sweep by measured block time --- extensions/tn_digest/internal/engine_ops.go | 45 +++++++- .../tn_digest/internal/prune_ops_test.go | 81 ++++++++++++++ extensions/tn_digest/scheduler/constants.go | 103 +++++++++--------- extensions/tn_digest/scheduler/scheduler.go | 22 ++++ 4 files changed, 196 insertions(+), 55 deletions(-) diff --git a/extensions/tn_digest/internal/engine_ops.go b/extensions/tn_digest/internal/engine_ops.go index ffb748da..5b7770fb 100644 --- a/extensions/tn_digest/internal/engine_ops.go +++ b/extensions/tn_digest/internal/engine_ops.go @@ -3,6 +3,7 @@ package internal import ( "context" "encoding/json" + "errors" "fmt" "strings" "time" @@ -14,6 +15,30 @@ import ( "github.com/trufnetwork/kwil-db/node/types/sql" ) +// ErrBroadcastPending reports that a broadcast returned without a committed +// result while the transaction is still live in the mempool. +// +// This is not a failure and must not be retried. ktypes.ErrTxTimeout means the +// wait for inclusion elapsed, not that the transaction was rejected -- it stays +// in the mempool and still executes. Retrying it with a fresh nonce therefore +// does not replace the work, it duplicates it, and the two transactions race +// for the same account's nonce sequence. +// +// Mainnet, 2026-09-08: auto_prune_duplicates executed 22 times on chain while +// the scheduler logged one success and five failures, because every timeout was +// retried. Each of those was a ~17 second consensus transaction, so the retry +// path roughly quadrupled the load during the window it was already hurting. +// +// A caller that sees this should stop its drain for the firing. The sweep is +// cyclic and resumes from its cursor, so nothing is lost by ending early. +var ErrBroadcastPending = errors.New("broadcast returned no committed result; transaction is still pending") + +// isBroadcastPending reports whether err means the transaction is live but +// uncommitted, rather than rejected. +func isBroadcastPending(err error) bool { + return errors.Is(err, ktypes.ErrTxTimeout) +} + // DigestTxResult represents the parsed result from an auto_digest transaction type DigestTxResult struct { ProcessedDays int @@ -325,7 +350,17 @@ func (e *EngineOperations) BroadcastAutoDigestWithArgsAndRetry( return result, nil } - // On ANY error, retry with fresh nonce after backoff + if isBroadcastPending(err) { + // See ErrBroadcastPending: still in the mempool, still going to execute. + // The digest path left the same nonce gaps as the prune one did. + e.logger.Warn("auto_digest broadcast did not confirm in time; leaving it pending rather than retrying", + "attempt", attempt, + "tx_hash", hash.String(), + "error", err) + return nil, fmt.Errorf("%w: %v", ErrBroadcastPending, err) + } + + // On any other error, retry with a fresh nonce after backoff lastErr = err e.logger.Warn("Broadcast failed, will retry with fresh nonce", "attempt", attempt, @@ -902,6 +937,14 @@ func (e *EngineOperations) BroadcastAutoPruneDuplicatesWithRetry( if err == nil { return result, nil } + if isBroadcastPending(err) { + // See ErrBroadcastPending: the transaction is still in the mempool and + // will execute. Retrying duplicates a ~17 second scan rather than + // replacing it. + e.logger.Warn("auto_prune_duplicates broadcast did not confirm in time; leaving it pending rather than retrying", + "attempt", attempt, "error", err) + return nil, fmt.Errorf("%w: %v", ErrBroadcastPending, err) + } lastErr = err } diff --git a/extensions/tn_digest/internal/prune_ops_test.go b/extensions/tn_digest/internal/prune_ops_test.go index 6eff78f7..a7254c55 100644 --- a/extensions/tn_digest/internal/prune_ops_test.go +++ b/extensions/tn_digest/internal/prune_ops_test.go @@ -3,6 +3,7 @@ package internal import ( "context" "errors" + "fmt" "testing" "time" @@ -85,6 +86,9 @@ func TestParsePruneResultFromTxLog_IgnoresTheDigestNotice(t *testing.T) { type prunePathBroadcaster struct { attempts int failUntil int + // failWith replaces the generic network error, so a test can distinguish a + // rejected transaction from one that is merely uncommitted. + failWith error // action and argCount record what the last transaction actually asked for. action string argCount int @@ -102,6 +106,9 @@ func (m *prunePathBroadcaster) broadcast(ctx context.Context, tx *ktypes.Transac result := &ktypes.TxResult{Code: uint32(ktypes.CodeOk), Log: pruneNotice} if m.attempts <= m.failUntil { + if m.failWith != nil { + return ktypes.Hash{}, result, m.failWith + } return ktypes.Hash{}, result, errors.New("network error") } return ktypes.Hash{1, 2, 3}, result, nil @@ -191,3 +198,77 @@ func TestBroadcastAutoPruneDuplicates_StopsOnContextCancellation(t *testing.T) { t.Fatalf("kept broadcasting past cancellation: %d attempts", broadcaster.attempts) } } + +// A broadcast timeout is not a rejection. The transaction stays in the mempool and +// still executes, so retrying it with a fresh nonce does not replace the work, it +// duplicates it -- and each duplicate is a ~17 second consensus transaction. +// +// This is the regression for the mainnet incident of 2026-09-08, where +// auto_prune_duplicates ran 22 times on chain while the scheduler believed it had +// managed one success and five failures. +func TestBroadcastAutoPruneDuplicates_DoesNotRetryAPendingTransaction(t *testing.T) { + accounts := &mockAccounts{} + // failUntil high enough that a retrying implementation would broadcast again. + broadcaster := &prunePathBroadcaster{failUntil: 100, failWith: ktypes.ErrTxTimeout} + ops := &EngineOperations{logger: log.New(), accounts: accounts} + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + _, err := ops.BroadcastAutoPruneDuplicatesWithRetry( + ctx, "test-chain", newPruneSigner(t), broadcaster.broadcast, 1000, 5, 3, + ) + if err == nil { + t.Fatal("expected an error when the broadcast does not confirm") + } + if !errors.Is(err, ErrBroadcastPending) { + t.Fatalf("want ErrBroadcastPending so the drain can end the firing, got %v", err) + } + if broadcaster.attempts != 1 { + t.Fatalf("a pending transaction was rebroadcast: want 1 attempt, got %d", broadcaster.attempts) + } +} + +// A wrapped timeout still has to be recognised: the broadcast path wraps with +// %w before the retry loop sees it. +func TestBroadcastAutoPruneDuplicates_RecognisesAWrappedTimeout(t *testing.T) { + accounts := &mockAccounts{} + broadcaster := &prunePathBroadcaster{ + failUntil: 100, + failWith: fmt.Errorf("broadcast tx: %w", ktypes.ErrTxTimeout), + } + ops := &EngineOperations{logger: log.New(), accounts: accounts} + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + _, err := ops.BroadcastAutoPruneDuplicatesWithRetry( + ctx, "test-chain", newPruneSigner(t), broadcaster.broadcast, 1000, 5, 3, + ) + if !errors.Is(err, ErrBroadcastPending) { + t.Fatalf("want ErrBroadcastPending for a wrapped timeout, got %v", err) + } + if broadcaster.attempts != 1 { + t.Fatalf("want 1 attempt, got %d", broadcaster.attempts) + } +} + +// The narrowing must not swallow ordinary failures: a rejected transaction is not +// pending and still deserves its retries. +func TestBroadcastAutoPruneDuplicates_StillRetriesARealFailure(t *testing.T) { + accounts := &mockAccounts{} + 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, 1000, 5, 3, + ); err != nil { + t.Fatalf("a network error should still be retried, got %v", err) + } + if broadcaster.attempts != 2 { + t.Fatalf("want 2 attempts, got %d", broadcaster.attempts) + } +} diff --git a/extensions/tn_digest/scheduler/constants.go b/extensions/tn_digest/scheduler/constants.go index 624b4c9a..1f4dead6 100644 --- a/extensions/tn_digest/scheduler/constants.go +++ b/extensions/tn_digest/scheduler/constants.go @@ -5,23 +5,21 @@ import "time" const ( // Digest drain mode constants (scheduler-scoped to avoid import cycles) // - // DigestDeleteCap bounds a single auto_digest transaction, and the bound that - // matters is not the row count on its own. kwild issues PREPARE TRANSACTION and - // then waits for Postgres logical replication to decode that block's whole - // change set and hand back a commit ID. That wait is a hardcoded 30 seconds - // (kwil-db node/pg/db.go). Exceeding it is fatal: the node exits, and because - // the block never commits, the same work is retried on the next firing and - // fails again. Testnet sat in exactly that loop for five consecutive digest - // firings, restarting each time and draining nothing. + // DigestDeleteCap bounds a single auto_digest transaction. auto_digest turns it + // into a day count as floor((delete_cap * 3) / (expected_records_per_stream * 2)), + // so 10,000 covers 625 days a run and matches the default the action itself + // declares. A backlog takes more runs, which DrainMaxRuns already allows for. // - // auto_digest turns this cap into a day count as - // floor((delete_cap * 3) / (expected_records_per_stream * 2)), so 100,000 asked - // for 6,250 days of history in one transaction. Runs that did commit were - // changing roughly 43,000 rows; the ones that killed the node were larger. At - // 10,000 a run covers 625 days and changes at most about 11,000 rows, which is - // comfortably inside the window, and it matches the default auto_digest itself - // declares. A backlog simply takes more runs, which DrainMaxRuns already allows - // for: 100 runs a firing still clears 62,500 days. + // An earlier revision of this comment blamed the testnet crash loop of 2026-09 + // on this cap being too large. That was wrong and is corrected here: the loop was + // log volume. Testnet ran at debug level on the awslogs driver, and the + // replication monitor emits one line per decoded row, shipped synchronously over + // the network -- roughly 2,300 lines a second. That blew the hardcoded 30 second + // precommit window in kwil-db node/pg/db.go. Setting the log level to info fixed + // it outright: 40 hours and six firings on the OLD 100,000 cap with no timeouts. + // The cap was reduced anyway and is kept here, because 625 days a run is a + // reasonable size on its own, but it was not the fix and should not be cited as + // one. DigestDeleteCap = 10_000 DigestExpectedRecordsPerStream = 24 DigestPreservePastDays = 2 @@ -57,52 +55,49 @@ const ( // // 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. + // rather than stopping early. // - // 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. - // Same 30-second replication window as DigestDeleteCap above, but this cap is in - // different units and cannot simply copy digest's number. batch_prune_duplicates - // bounds EVENT TIMES, not rows (057 says why: rule 5 makes an event time atomic - // and primitive_events has no primary key to address a single row by), and one - // event time expands into the change set twice over: + // These numbers come from what the first mainnet firing actually did, on + // 2026-09-08. At PruneStreamBatchSize 100 a single auto_prune_duplicates + // transaction held the consensus path for a median of 16.9 seconds (22 samples, + // max 18.6 s) against neighbouring blocks at 50 ms. The gateway's read timeout + // is 20 s, so user reads queued behind it and 1,619 of them returned 503 with + // rpc_code -32001 across a 19 minute window. // - // - every revision at it, since the whole event time goes together - // - every marker in that event time's whole DAY, because a digested day that - // kept some markers and lost others reads as corruption, so step 3 clears - // the day rather than the one marker + // The lesson is which knob matters. Run 1 deleted 794 event times in 14.4 s; + // later runs deleted essentially nothing and still cost 16-18 s. The work is the + // per-stream history scan across stream_batch_size streams, and PruneDeleteCap + // bounds deletions only -- it cannot bound the scan, so lowering it does nothing + // for block time. Size PruneStreamBatchSize against measured block time. // - // Measured on mainnet (2% page samples, 2026-09-06): rows per event time average - // 1.0001 and top out at 2, with 84 of 1,486,639 sampled event times carrying any - // revision at all. Markers per stream-day average 1.0904 and top out at 4, which - // is also the structural ceiling since digest writes at most open/high/low/close. + // A whole pass costs what it costs: ~182,000 primitive streams at 17 s per 100 + // streams is about 8.6 hours of execution however it is sliced. Batching only + // decides how that is spread, so the shape is many cheap transactions at a low + // duty cycle rather than a few expensive ones. // - // So a cap of C changes at most 2C rows plus 4C markers, 6C in the worst case and - // about 2.1C typically. Duplicate-heavy streams are daily publishers whose days - // hold one marker each, so they sit at the low end, but the cap has to hold for - // the intraday streams too. At 5,000 that is 30,000 changes worst case and around - // 10,500 in practice, which matches what DigestDeleteCap allows. 10,000 would - // have reached 60,000, past the ~43,000 digest was still committing at before it - // stopped fitting the window. + // At 5 streams a run a transaction lands near 0.85 s, which is an ordinary block + // rather than 85% of the read budget. One run per 10 s is an ~8% duty cycle, and + // 1,000 runs covers 5,000 streams in about 2.8 hours -- inside the six-hourly + // firing, with a full pass in roughly 37 firings, about nine days. // - // batch_prune_duplicates returns deleted_rows next to deleted_event_times, so the - // real ratio is observable per run. Raise this from that measurement, not from a - // guess. - PruneDeleteCap = 5_000 - PruneStreamBatchSize = 100 - PruneDrainMaxRuns = 100 + // What this does NOT fix: the per-stream scan is unbounded, because the deletable + // set is a whole-stream property. One stream holding 1.3 M rows costs the same in + // a batch of 5 as in a batch of 100. That tail needs a design change -- a cursor + // within a stream, or a maintained duplicate index -- not a smaller constant. + // Re-measure with internal/benchmark/digest on a mainnet-shaped fixture before + // raising any of these; testnet is 212k rows at 1.8% duplicates and never reaches + // this path. + PruneDeleteCap = 1_000 + PruneStreamBatchSize = 5 + PruneDrainMaxRuns = 1_000 - // PruneDrainRunDelay paces the runs that actually delete, the way digest's - // DrainRunDelay paces its own capped deletes. - PruneDrainRunDelay = 60 * time.Second + // PruneDrainRunDelay paces the runs that actually delete. Ten seconds against a + // ~0.85 s transaction is the duty cycle described above; the old 60 s was chosen + // when a run was rare and expensive rather than frequent and cheap. + PruneDrainRunDelay = 10 * 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. + // cycle -- so this stays short. PruneIdleRunDelay = 5 * time.Second PruneDrainMaxConsecutiveFailures = 5 ) diff --git a/extensions/tn_digest/scheduler/scheduler.go b/extensions/tn_digest/scheduler/scheduler.go index 2983fe12..32fc3a5c 100644 --- a/extensions/tn_digest/scheduler/scheduler.go +++ b/extensions/tn_digest/scheduler/scheduler.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "errors" "fmt" "runtime/debug" "sync" @@ -157,6 +158,15 @@ func (s *DigestScheduler) Start(ctx context.Context, cronExpr string) error { ) if err != nil { + if errors.Is(err, internal.ErrBroadcastPending) { + // Same reasoning as the prune drain: the transaction is live and + // will execute, so ending the firing is cheaper than stacking on it. + s.logger.Info("digest drain ending early: a broadcast is still pending", + "runs_completed", runs, + "error", err) + return + } + consecutiveFailures++ s.logger.Warn("auto_digest broadcast failed after retries", "run", runs, @@ -453,6 +463,18 @@ func (s *DigestScheduler) runPruneDrain(ctx context.Context) { delay := PruneIdleRunDelay if err != nil { + if errors.Is(err, internal.ErrBroadcastPending) { + // The transaction is still in the mempool and will execute. Carrying + // on would stack a second scan on top of it and race its nonce, so + // the firing ends here; the cursor resumes it next time. + s.logger.Info("duplicate prune drain ending early: a broadcast is still pending", + "runs_completed", runs, + "cumulative_swept", totalSweptStreams, + "cumulative_deleted_rows", totalRows, + "error", err) + return + } + consecutiveFailures++ s.logger.Warn("auto_prune_duplicates broadcast failed after retries", "run", runs,