chore: bound node transaction-event ledger growth

extensions/tn_digest/engine_ops_integration_test.go

@@ -4,6 +4,8 @@ package tn_digest
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -16,6 +18,7 @@ import (
 	"github.com/trufnetwork/kwil-db/node/accounts"
 	kwilTesting "github.com/trufnetwork/kwil-db/testing"
 	"github.com/trufnetwork/node/extensions/tn_digest/internal"
+	"github.com/trufnetwork/node/internal/migrations"
 	digestembed "github.com/trufnetwork/node/tests/extensions/digest"
 	testutils "github.com/trufnetwork/node/tests/streams/utils"
 )
@@ -92,3 +95,105 @@ func TestBuildAndBroadcastAutoDigestTx_VerifiesTxBuildSignAndDBEffect(t *testing
 		},
 	}, nil)
 }
+
+// TestBroadcastTrimTransactionEventsTx_VerifiesTxBuildSignAndDBEffect
+// Mirrors the auto_digest broadcast test for the retention path:
+//   - seeds the real schema and one old high-volume write-fee row (method 2)
+//   - builds and signs an ActionExecution tx for trim_transaction_events via the engine-op
+//   - the broadcaster stub verifies the payload, executes the action above the
+//     cutoff (authz overridden), and returns a NOTICE the parser understands
+//   - asserts the parsed result and that the old row was pruned from the DB
+func TestBroadcastTrimTransactionEventsTx_VerifiesTxBuildSignAndDBEffect(t *testing.T) {
+	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
+		Name:           "tn_digest_trim_tx_events_broadcast_test",
+		SeedStatements: migrations.GetSeedScriptStatements(),
+		FunctionTests: []kwilTesting.TestFunc{
+			func(ctx context.Context, platform *kwilTesting.Platform) error {
+				accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New())
+				require.NoError(t, err)
+
+				ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, accts, log.New())
+
+				priv, _, err := crypto.GenerateSecp256k1Key(nil)
+				require.NoError(t, err)
+				signer := auth.GetNodeSigner(priv)
+				require.NotNil(t, signer)
+
+				// Seed one old high-volume write-fee row (method 2) to be pruned.
+				oldTxID := fmt.Sprintf("0x%064x", 0xAA)
+				_, err = platform.DB.Execute(ctx,
+					`INSERT INTO main.transaction_events (tx_id, block_height, method_id, caller, fee_amount, fee_recipient, metadata)
+					 VALUES ($1, $2, $3, $4, $5::NUMERIC(78, 0), $6, NULL)`,
+					oldTxID, int64(100), int64(2), "0x9999999999999999999999999999999999999999",
+					"1000000000000000000", "0x1111111111111111111111111111111111111111")
+				require.NoError(t, err)
+
+				const preserveBlocks = int64(172_800)
+				const deleteCap = 100
+
+				// The broadcaster executes the REAL action at trimHeight (authz
+				// overridden, since the node signer is not the namespace owner) and
+				// returns the action's own NOTICE — so the engine-op parses the real
+				// migration output, not a fabricated log.
+				var trimHeight int64
+				broadcaster := func(ctx context.Context, tx *types.Transaction, sync uint8) (types.Hash, *types.TxResult, error) {
+					payload, err := types.UnmarshalPayload(tx.Body.PayloadType, tx.Body.Payload)
+					require.NoError(t, err)
+					ae, ok := payload.(*types.ActionExecution)
+					require.True(t, ok, "payload should be ActionExecution")
+					require.Equal(t, "main", ae.Namespace)
+					require.Equal(t, "trim_transaction_events", ae.Action)
+					require.Len(t, ae.Arguments, 1)
+					require.Len(t, ae.Arguments[0], 2, "trim takes preserve_blocks + delete_cap")
+
+					txCtx := &common.TxContext{
+						Ctx:          ctx,
+						BlockContext: &common.BlockContext{Height: trimHeight},
+						Signer:       signer.CompactID(),
+						Caller:       "node",
+						TxID:         "test-tx",
+					}
+					engCtx := &common.EngineContext{TxContext: txCtx, OverrideAuthz: true}
+					res, execErr := platform.Engine.Call(engCtx, platform.DB, "main", "trim_transaction_events",
+						[]any{preserveBlocks, int64(deleteCap)}, func(_ *common.Row) error { return nil })
+					require.NoError(t, execErr)
+					require.NoError(t, res.Error)
+
+					return types.Hash{}, &types.TxResult{
+						Code: uint32(types.CodeOk),
+						Log:  strings.Join(res.Logs, "\n"),
+					}, nil
+				}
+
+				// Case 1: height above the cutoff (200000 - 172800 = 27200) prunes
+				// the seeded write-fee row (and its cascade child).
+				trimHeight = 200_000
+				ctx1, cancel1 := context.WithTimeout(ctx, 10*time.Second)
+				defer cancel1()
+				result, err := ops.BroadcastTrimTransactionEventsWithRetry(ctx1, "tn-test", signer, broadcaster, preserveBlocks, deleteCap, 3)
+				require.NoError(t, err)
+				require.Equal(t, 1, result.Deleted)
+				require.False(t, result.HasMore)
+
+				// Verify DB side-effect: the old write-fee row is gone.
+				sel, qErr := platform.DB.Execute(ctx, `SELECT 1 FROM main.transaction_events WHERE tx_id = $1`, oldTxID)
+				require.NoError(t, qErr)
+				require.Empty(t, sel.Rows, "old write-fee row should have been pruned")
+
+				// Case 2: chain younger than the preserve window (cutoff <= 0) must
+				// report a clean no-op through the broadcast path — a parseable
+				// NOTICE, no retries, no error. (Regression guard for the cutoff<=0
+				// NOTICE format in migration 052.)
+				trimHeight = 100
+				ctx3, cancel3 := context.WithTimeout(ctx, 10*time.Second)
+				defer cancel3()
+				noop, err := ops.BroadcastTrimTransactionEventsWithRetry(ctx3, "tn-test", signer, broadcaster, preserveBlocks, deleteCap, 3)
+				require.NoError(t, err, "cutoff<=0 no-op must parse cleanly without retries")
+				require.Equal(t, 0, noop.Deleted)
+				require.False(t, noop.HasMore)
+
+				return nil
+			},
+		},
+	}, testutils.GetTestOptionsWithCache())
+}

extensions/tn_digest/internal/engine_ops.go

@@ -601,6 +601,165 @@ func parseTrimResultFromTxLog(logOutput string) (*TrimOrderEventsTxResult, error
 	return nil, fmt.Errorf("no trim_order_events log entry found in: %q", logOutput)
 }
 
+// TrimTransactionEventsTxResult represents the parsed result from a
+// trim_transaction_events transaction.
+type TrimTransactionEventsTxResult struct {
+	Deleted   int
+	Remaining int
+	HasMore   bool
+}
+
+// BroadcastTrimTransactionEventsWithRetry wraps broadcastTrimTransactionEventsOnce
+// with retry logic. On any error it re-fetches a fresh nonce before retrying,
+// mirroring BroadcastTrimOrderEventsWithRetry.
+func (e *EngineOperations) BroadcastTrimTransactionEventsWithRetry(
+	ctx context.Context,
+	chainID string,
+	signer auth.Signer,
+	broadcaster func(context.Context, *ktypes.Transaction, uint8) (ktypes.Hash, *ktypes.TxResult, error),
+	preserveBlocks int64,
+	deleteCap int,
+	maxRetries int,
+) (*TrimTransactionEventsTxResult, 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 trim_transaction_events 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.broadcastTrimTransactionEventsOnce(ctx, chainID, signer, broadcaster, preserveBlocks, deleteCap)
+		if err == nil {
+			return result, nil
+		}
+		lastErr = err
+	}
+
+	return nil, fmt.Errorf("trim_transaction_events max retries (%d) exceeded: %w", maxRetries, lastErr)
+}
+
+// broadcastTrimTransactionEventsOnce fetches a fresh nonce and broadcasts once.
+func (e *EngineOperations) broadcastTrimTransactionEventsOnce(
+	ctx context.Context,
+	chainID string,
+	signer auth.Signer,
+	broadcaster func(context.Context, *ktypes.Transaction, uint8) (ktypes.Hash, *ktypes.TxResult, error),
+	preserveBlocks int64,
+	deleteCap int,
+) (*TrimTransactionEventsTxResult, 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)
+	}
+
+	preserveBlocksArg, err := ktypes.EncodeValue(preserveBlocks)
+	if err != nil {
+		return nil, fmt.Errorf("encode preserveBlocks: %w", err)
+	}
+	deleteCapArg, err := ktypes.EncodeValue(int64(deleteCap))
+	if err != nil {
+		return nil, fmt.Errorf("encode deleteCap: %w", err)
+	}
+
+	payload := &ktypes.ActionExecution{
+		Namespace: "main",
+		Action:    "trim_transaction_events",
+		Arguments: [][]*ktypes.EncodedValue{{
+			preserveBlocksArg, deleteCapArg,
+		}},
+	}
+
+	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 := parseTxEventsTrimResultFromTxLog(txResult.Log)
+	if err != nil {
+		e.logger.Warn("failed to parse trim_transaction_events result", "error", err, "log", txResult.Log)
+		return nil, fmt.Errorf("parse trim result: %w", err)
+	}
+
+	e.logger.Info("trim_transaction_events completed",
+		"deleted", result.Deleted,
+		"remaining", result.Remaining,
+		"has_more", result.HasMore,
+		"tx_hash", hash.String())
+
+	return result, nil
+}
+
+// parseTxEventsTrimResultFromTxLog extracts deleted/remaining/has_more from NOTICE
+// output, mirroring parseTrimResultFromTxLog but keyed on the
+// trim_transaction_events marker.
+func parseTxEventsTrimResultFromTxLog(logOutput string) (*TrimTransactionEventsTxResult, error) {
+	result := &TrimTransactionEventsTxResult{}
+	for _, line := range strings.Split(logOutput, "\n") {
+		if !strings.Contains(line, "trim_transaction_events:") {
+			continue
+		}
+		// Extract payload after marker (handles prefixed lines like "NOTICE: trim_transaction_events: ...")
+		parts := strings.SplitN(line, "trim_transaction_events:", 2)
+		if len(parts) != 2 {
+			continue
+		}
+		payload := strings.TrimSpace(parts[1])
+
+		n, err := fmt.Sscanf(payload, "deleted=%d remaining=%d has_more=%t",
+			&result.Deleted, &result.Remaining, &result.HasMore)
+		if err != nil || n != 3 {
+			return nil, fmt.Errorf("failed to parse trim log payload (scanned %d/3 fields): %q", n, payload)
+		}
+		return result, nil
+	}
+	return nil, fmt.Errorf("no trim_transaction_events log entry found in: %q", logOutput)
+}
+
 // parseDigestResultFromTxLog parses the digest result from transaction log output
 // The digest action outputs log entries like: "auto_digest:{...json...}"
 func parseDigestResultFromTxLog(logOutput string) (*DigestTxResult, error) {

extensions/tn_digest/internal/engine_ops_test.go

@@ -57,6 +57,68 @@ func TestParseDigestResultFromTxLog_NoEntry(t *testing.T) {
 	}
 }
 
+func TestParseTxEventsTrimResultFromTxLog_HasMoreTrue(t *testing.T) {
+	log := "INFO something\nNOTICE: trim_transaction_events: deleted=2 remaining=5 has_more=true\nother"
+	res, err := parseTxEventsTrimResultFromTxLog(log)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res.Deleted != 2 {
+		t.Fatalf("deleted: want 2, got %d", res.Deleted)
+	}
+	if res.Remaining != 5 {
+		t.Fatalf("remaining: want 5, got %d", res.Remaining)
+	}
+	if !res.HasMore {
+		t.Fatalf("has_more: want true, got false")
+	}
+}
+
+func TestParseTxEventsTrimResultFromTxLog_HasMoreFalse(t *testing.T) {
+	log := "trim_transaction_events: deleted=3 remaining=0 has_more=false"
+	res, err := parseTxEventsTrimResultFromTxLog(log)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res.Deleted != 3 {
+		t.Fatalf("deleted: want 3, got %d", res.Deleted)
+	}
+	if res.Remaining != 0 {
+		t.Fatalf("remaining: want 0, got %d", res.Remaining)
+	}
+	if res.HasMore {
+		t.Fatalf("has_more: want false, got true")
+	}
+}
+
+func TestParseTxEventsTrimResultFromTxLog_NoEntry(t *testing.T) {
+	log := "INFO: nothing relevant here\nNOTICE: trim_order_events: deleted=1 remaining=0 has_more=false"
+	_, err := parseTxEventsTrimResultFromTxLog(log)
+	if err == nil {
+		t.Fatalf("expected error for missing trim_transaction_events entry, got nil")
+	}
+}
+
+// TestParseTxEventsTrimResultFromTxLog_CutoffNoop asserts the exact NOTICE the
+// migration emits on the cutoff<=0 branch parses as a clean no-op (no error, no
+// retries). Guards the parser<->migration contract for the young-chain case.
+func TestParseTxEventsTrimResultFromTxLog_CutoffNoop(t *testing.T) {
+	log := "NOTICE: trim_transaction_events: deleted=0 remaining=0 has_more=false"
+	res, err := parseTxEventsTrimResultFromTxLog(log)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res.Deleted != 0 {
+		t.Fatalf("deleted: want 0, got %d", res.Deleted)
+	}
+	if res.Remaining != 0 {
+		t.Fatalf("remaining: want 0, got %d", res.Remaining)
+	}
+	if res.HasMore {
+		t.Fatalf("has_more: want false, got true")
+	}
+}
+
 // Mock implementations for testing retry logic
 
 type mockBroadcaster struct {

extensions/tn_digest/scheduler/constants.go

@@ -16,4 +16,17 @@ const (
 	TrimOrderEventsPreserveBlocks int64 = 172_800
 	TrimOrderEventsDeleteCap            = 100_000
 	TrimOrderEventsMaxRuns              = 10
+
+	// Transaction event trim constants
+	// ~2 days at 1-second blocks, so the indexer has synced high-volume
+	// write-fee (method 2) ledger rows before they are pruned from node state.
+	TrimTxEventsPreserveBlocks int64 = 172_800
+	TrimTxEventsDeleteCap      int   = 100_000
+	TrimTxEventsMaxRuns        int   = 10
+
+	// TrimTxEventsEnabled gates activation. It ships false so a binary rollout
+	// is decoupled from actually pruning: enable only after the Trufscan
+	// indexer fallback (trufscan #183) is live in prod, so a pruned tx still
+	// resolves on the explorer /tx page.
+	TrimTxEventsEnabled bool = false
 )