chore: pace the duplicate prune sweep by measured block time

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
 	}
 

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

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
 )

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,