diff --git a/extensions/tn_digest/engine_ops_integration_test.go b/extensions/tn_digest/engine_ops_integration_test.go index 5cbf818b..3c7b459e 100644 --- a/extensions/tn_digest/engine_ops_integration_test.go +++ b/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()) +} diff --git a/extensions/tn_digest/internal/engine_ops.go b/extensions/tn_digest/internal/engine_ops.go index 9722dbf3..64b15762 100644 --- a/extensions/tn_digest/internal/engine_ops.go +++ b/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) { diff --git a/extensions/tn_digest/internal/engine_ops_test.go b/extensions/tn_digest/internal/engine_ops_test.go index a801a33a..ab36ebbe 100644 --- a/extensions/tn_digest/internal/engine_ops_test.go +++ b/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 { diff --git a/extensions/tn_digest/scheduler/constants.go b/extensions/tn_digest/scheduler/constants.go index 76263e3a..f258c495 100644 --- a/extensions/tn_digest/scheduler/constants.go +++ b/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 ) diff --git a/extensions/tn_digest/scheduler/scheduler.go b/extensions/tn_digest/scheduler/scheduler.go index dfae36f5..1d9639b4 100644 --- a/extensions/tn_digest/scheduler/scheduler.go +++ b/extensions/tn_digest/scheduler/scheduler.go @@ -168,6 +168,8 @@ func (s *DigestScheduler) Start(ctx context.Context, cronExpr string) error { // After digest drain, trim order events (best-effort, non-fatal) s.trimOrderEvents(jobCtx, chainID, engineOps, signer, broadcaster) + // Then trim high-volume transaction-event ledger rows (best-effort, non-fatal) + s.trimTransactionEvents(jobCtx, chainID, engineOps, signer, broadcaster) return } } @@ -188,6 +190,8 @@ func (s *DigestScheduler) Start(ctx context.Context, cronExpr string) error { // After digest drain, trim order events (best-effort, non-fatal) s.trimOrderEvents(jobCtx, chainID, engineOps, signer, broadcaster) + // Then trim high-volume transaction-event ledger rows (best-effort, non-fatal) + s.trimTransactionEvents(jobCtx, chainID, engineOps, signer, broadcaster) } if j, err := s.cron.Cron(cronExpr).Do(jobFunc); err != nil { @@ -269,6 +273,64 @@ func (s *DigestScheduler) trimOrderEvents( s.logger.Info("order event trim reached max runs", "max_runs", TrimOrderEventsMaxRuns) } +// trimTransactionEvents runs the trim_transaction_events action in a drain loop +// (best-effort). Called after the order-event trim completes. Gated by +// TrimTxEventsEnabled so pruning stays off until deliberately activated (after +// the Trufscan indexer fallback is live). Failures are logged but do not fail +// the digest job. +func (s *DigestScheduler) trimTransactionEvents( + ctx context.Context, + chainID string, + engineOps *internal.EngineOperations, + signer auth.Signer, + broadcaster txBroadcaster, +) { + if !TrimTxEventsEnabled { + return + } + + s.logger.Info("starting transaction event trim") + + for run := 0; run < TrimTxEventsMaxRuns; run++ { + select { + case <-ctx.Done(): + s.logger.Info("transaction event trim canceled", "runs_completed", run) + return + default: + } + + result, err := engineOps.BroadcastTrimTransactionEventsWithRetry( + ctx, chainID, signer, broadcaster.BroadcastTx, + TrimTxEventsPreserveBlocks, + TrimTxEventsDeleteCap, + 3, // maxRetries + ) + if err != nil { + s.logger.Warn("trim_transaction_events failed (non-fatal)", "run", run, "error", err) + return + } + + s.logger.Info("trim_transaction_events run completed", + "run", run, + "deleted", result.Deleted, + "remaining", result.Remaining, + "has_more", result.HasMore) + + if !result.HasMore { + return + } + + // Brief delay between trim runs + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + } + + s.logger.Info("transaction event trim reached max runs", "max_runs", TrimTxEventsMaxRuns) +} + // RunOnce executes the digest job payload once (for tests and manual triggering). func (s *DigestScheduler) RunOnce(ctx context.Context) error { if s.engineOps == nil || s.broadcaster == nil || s.signer == nil || s.kwilService == nil || s.kwilService.GenesisConfig == nil { diff --git a/internal/migrations/052-transaction-events-retention.sql b/internal/migrations/052-transaction-events-retention.sql new file mode 100644 index 00000000..2bb18ac0 --- /dev/null +++ b/internal/migrations/052-transaction-events-retention.sql @@ -0,0 +1,94 @@ +/** + * ============================================================================= + * Transaction Ledger Retention + * ============================================================================= + * + * Bounds growth of main.transaction_events (and its CASCADE child + * transaction_event_distributions) — a consensus-state table that otherwise + * accumulates a row per fee event forever, replicated to every validator and + * captured in snapshots/statesync. + * + * Only the high-volume write-fee rows are trimmed: method_id = 2 (insertRecords, + * written by both insert_records and the truflation insert path). These are a + * flat 1 TRUF -> leader per write, so the per-tx ledger row is uniform and the + * transaction itself remains permanently resolvable off-chain via the Trufscan + * chain indexer. Low-volume, semantically-rich classes (deployStream 1, + * setTaxonomies 3, requestAttestation 6, setMetadata 7, createMarket 8) are left + * untouched so their fee/distribution history is preserved. + * + * Follows the same write -> index -> trim pattern as trim_order_events (044), + * driven by the leader-only tn_digest scheduler. Deletion is deterministic + * (block-height cutoff, never wall-clock) and delete-capped, so every validator + * removes exactly the same rows when replaying the block. + * + * Sequencing: the Trufscan indexer fallback (trufscan #183) must be live before + * this is activated, so that a pruned tx still resolves on the explorer /tx page. + */ + +-- ============================================================================= +-- trim_transaction_events: delete old high-volume write-fee ledger rows +-- ============================================================================= +/** + * Called by the tn_digest scheduler (leader-only) to trim old transaction + * events. Uses a block_height cutoff with a configurable buffer so the indexer + * has had time to sync before deletion. The CASCADE on + * transaction_event_distributions removes each trimmed row's fee breakdown. + * + * Parameters: + * - $preserve_blocks: Number of recent blocks to keep (e.g., 172800 = ~2 days) + * - $delete_cap: Maximum rows to delete per invocation (prevents large txs) + * + * Returns via NOTICE: deleted count, remaining count, has_more flag + */ +CREATE OR REPLACE ACTION trim_transaction_events( + $preserve_blocks INT8, + $delete_cap INT +) PUBLIC owner { + -- Only the high-volume write-fee ledger rows are trimmed; other methods + -- carry irreplaceable fee/distribution history and are kept. + $write_method_id INT := 2; -- insertRecords (insert_records + truflation) + + $cutoff INT8 := @height - $preserve_blocks; + + -- Don't trim if cutoff is negative (chain is younger than preserve window). + -- Emit the parseable no-op form so the scheduler's broadcast path reads a + -- clean success (deleted=0, has_more=false) instead of an unrecognized NOTICE + -- that would look like a parse failure and trigger needless retries. + if $cutoff <= 0 { + NOTICE('trim_transaction_events: deleted=0 remaining=0 has_more=false'); + RETURN; + } + + $count INT; + for $row in SELECT count(*)::INT as cnt FROM transaction_events + WHERE method_id = $write_method_id AND block_height < $cutoff { + $count := $row.cnt; + } + + if $count = 0 { + NOTICE('trim_transaction_events: deleted=0 remaining=0 has_more=false'); + RETURN; + } + + $to_delete INT; + if $count > $delete_cap { + $to_delete := $delete_cap; + } else { + $to_delete := $count; + } + + DELETE FROM transaction_events + WHERE tx_id IN ( + SELECT tx_id FROM transaction_events + WHERE method_id = $write_method_id + AND block_height < $cutoff + ORDER BY block_height ASC, tx_id ASC + LIMIT $to_delete + ); + + $remaining INT := $count - $to_delete; + $has_more BOOL := $count > $delete_cap; + NOTICE('trim_transaction_events: deleted=' || $to_delete::TEXT + || ' remaining=' || $remaining::TEXT + || ' has_more=' || $has_more::TEXT); +}; diff --git a/tests/streams/transaction_events_ledger_test.go b/tests/streams/transaction_events_ledger_test.go index f02a2389..ece358d9 100644 --- a/tests/streams/transaction_events_ledger_test.go +++ b/tests/streams/transaction_events_ledger_test.go @@ -895,3 +895,238 @@ func runTransactionIDTrackingScenario(t *testing.T) func(ctx context.Context, pl return nil } } + +const ( + trimTestCaller = "0x9999999999999999999999999999999999999999" + trimTestRecipient = "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +) + +// TestTrimTransactionEvents verifies the retention action from migration 052: +// only high-volume write-fee rows (method 2) past the block-height cutoff are +// pruned, low-volume fee-bearing classes are kept, the delete cap is honored, +// distribution children cascade, and a young chain is a no-op. +func TestTrimTransactionEvents(t *testing.T) { + owner := util.Unsafe_NewEthereumAddressFromString("0x1111111111111111111111111111111111111111") + testutils.RunSchemaTest(t, kwilTesting.SchemaTest{ + Name: "LEDGER_TRIM01_TransactionEventsRetention", + SeedStatements: migrations.GetSeedScriptStatements(), + Owner: owner.Address(), + FunctionTests: []kwilTesting.TestFunc{ + runTrimTransactionEventsScenario(t, owner), + }, + }, testutils.GetTestOptionsWithCache()) +} + +func runTrimTransactionEventsScenario(t *testing.T, owner util.EthereumAddress) func(ctx context.Context, platform *kwilTesting.Platform) error { + return func(ctx context.Context, platform *kwilTesting.Platform) error { + platform.Deployer = owner.Bytes() + + const ( + preserveBlocks = int64(172_800) + trimHeight = int64(200_000) // cutoff = 200000 - 172800 = 27200 + cutoff = int64(27_200) + methodInsert = 2 // insertRecords — high-volume write fee (trimmed) + methodDeploy = 1 // deployStream (kept) + methodAttest = 6 // requestAttestation (kept) + ) + + // --- Phase A: selectivity + cascade --- + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(1), 100, methodInsert, true)) // old write, with distribution child + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(2), 200, methodInsert, false)) // old write + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(3), 190_000, methodInsert, false)) // recent write (>= cutoff) + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(4), 100, methodDeploy, false)) // old deploy (kept) + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(5), 100, methodAttest, false)) // old attestation (kept) + + require.NoError(t, callTrimTxEvents(ctx, platform, owner, trimHeight, preserveBlocks, 100)) + + requireTxEvent(t, ctx, platform, trimTxHash(1), false) // old write pruned + requireTxEvent(t, ctx, platform, trimTxHash(2), false) // old write pruned + requireTxEvent(t, ctx, platform, trimTxHash(3), true) // recent write kept + requireTxEvent(t, ctx, platform, trimTxHash(4), true) // deploy kept (wrong method) + requireTxEvent(t, ctx, platform, trimTxHash(5), true) // attestation kept (wrong method) + requireDistribution(t, ctx, platform, trimTxHash(1), false) // child cascade-deleted with parent + + // --- Phase B: delete cap leaves a drainable remainder --- + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(10), 101, methodInsert, false)) + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(11), 102, methodInsert, false)) + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(12), 103, methodInsert, false)) + require.NoError(t, callTrimTxEvents(ctx, platform, owner, trimHeight, preserveBlocks, 2)) + require.Equal(t, 1, countTxEvents(t, ctx, platform, methodInsert, cutoff), + "delete_cap should leave exactly one old write-fee row for the next drain run") + + // --- Phase C: cutoff <= 0 (young chain) is a no-op --- + require.NoError(t, seedTxEvent(ctx, platform, trimTxHash(20), 50, methodInsert, false)) + require.NoError(t, callTrimTxEvents(ctx, platform, owner, 10, preserveBlocks, 100)) // cutoff = 10-172800 < 0 + requireTxEvent(t, ctx, platform, trimTxHash(20), true) // untouched + + return nil + } +} + +// trimTxHash returns a deterministic, well-formed 0x-prefixed 32-byte tx hash. +func trimTxHash(n int) string { + return fmt.Sprintf("0x%064x", n) +} + +// seedTxEvent inserts a transaction_events row directly (bypassing +// record_transaction_event's validation) so the trim can be exercised over a +// controlled mix of methods/heights. When withDistribution is set, it also seeds +// the CASCADE child so cascade behavior can be asserted. +func seedTxEvent(ctx context.Context, platform *kwilTesting.Platform, txID string, height int64, methodID int, withDistribution bool) error { + _, 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)`, + txID, height, int64(methodID), trimTestCaller, feeOneTRUF, trimTestRecipient) + if err != nil { + return err + } + if withDistribution { + _, err = platform.DB.Execute(ctx, + `INSERT INTO main.transaction_event_distributions (tx_id, sequence, recipient, amount, note) + VALUES ($1, 1, $2, $3::NUMERIC(78, 0), NULL)`, + txID, trimTestRecipient, feeOneTRUF) + } + return err +} + +func callTrimTxEvents(ctx context.Context, platform *kwilTesting.Platform, owner util.EthereumAddress, height, preserveBlocks, deleteCap int64) error { + tx := &common.TxContext{ + Ctx: ctx, + BlockContext: &common.BlockContext{Height: height}, + Signer: owner.Bytes(), + Caller: owner.Address(), + TxID: platform.Txid(), + Authenticator: coreauth.EthPersonalSignAuth, + } + engineCtx := &common.EngineContext{TxContext: tx} + res, err := platform.Engine.Call(engineCtx, platform.DB, "", "trim_transaction_events", + []any{preserveBlocks, deleteCap}, func(row *common.Row) error { return nil }) + if err != nil { + return err + } + if res != nil && res.Error != nil { + return res.Error + } + return nil +} + +func requireTxEvent(t *testing.T, ctx context.Context, platform *kwilTesting.Platform, txID string, wantExists bool) { + t.Helper() + res, err := platform.DB.Execute(ctx, `SELECT 1 FROM main.transaction_events WHERE tx_id = $1`, txID) + require.NoError(t, err) + require.Equal(t, wantExists, len(res.Rows) > 0, "transaction_events row existence for %s", txID) +} + +func requireDistribution(t *testing.T, ctx context.Context, platform *kwilTesting.Platform, txID string, wantExists bool) { + t.Helper() + res, err := platform.DB.Execute(ctx, `SELECT 1 FROM main.transaction_event_distributions WHERE tx_id = $1`, txID) + require.NoError(t, err) + require.Equal(t, wantExists, len(res.Rows) > 0, "transaction_event_distributions row existence for %s", txID) +} + +func countTxEvents(t *testing.T, ctx context.Context, platform *kwilTesting.Platform, methodID int, cutoff int64) int { + t.Helper() + res, err := platform.DB.Execute(ctx, + `SELECT count(*)::INT8 FROM main.transaction_events WHERE method_id = $1 AND block_height < $2`, + int64(methodID), cutoff) + require.NoError(t, err) + require.Len(t, res.Rows, 1) + return int(res.Rows[0][0].(int64)) +} + +// TestTrimTransactionEventsE2E drives the retention over rows produced by the +// REAL write path: genuine create_streams (deployStream, method 1) and +// insert_records (insertRecords, method 2) txs record ledger rows via +// record_transaction_event, then trim_transaction_events prunes only the old +// high-volume method-2 row (and its cascade child), keeping the recent write and +// the deployStream row. +func TestTrimTransactionEventsE2E(t *testing.T) { + owner := util.Unsafe_NewEthereumAddressFromString("0x7E5F4552091A69125d5DfCb7b8C2659029395Bdf") + testutils.RunSchemaTest(t, kwilTesting.SchemaTest{ + Name: "LEDGER_TRIM02_TransactionEventsRetentionE2E", + SeedStatements: migrations.GetSeedScriptStatements(), + Owner: owner.Address(), + FunctionTests: []kwilTesting.TestFunc{ + runTrimTxEventsE2EScenario(t, owner), + }, + }, testutils.GetTestOptionsWithCache()) +} + +func runTrimTxEventsE2EScenario(t *testing.T, owner util.EthereumAddress) func(ctx context.Context, platform *kwilTesting.Platform) error { + return func(ctx context.Context, platform *kwilTesting.Platform) error { + platform.Deployer = owner.Bytes() + + require.NoError(t, setup.AddMemberToRoleBypass(ctx, platform, "system", "network_writers_manager", owner.Address())) + require.NoError(t, setup.CreateDataProvider(ctx, platform, owner.Address())) + + err := erc20shim.ForTestingSeedAndActivateInstance(ctx, platform, ledgerChain, ledgerEscrow, ledgerERC20, 18, 60, ledgerExtensionAlias) + if err != nil { + if !strings.Contains(err.Error(), "alias \"sepolia_bridge\" already exists") { + require.NoError(t, err) + } + require.NoError(t, erc20shim.ForTestingInitializeExtension(ctx, platform)) + } + require.NoError(t, erc20shim.ForTestingInitializeExtension(ctx, platform)) + + actorVal := util.Unsafe_NewEthereumAddressFromString("0x9999999999999999999999999999999999999999") + actor := &actorVal + require.NoError(t, setup.CreateDataProviderWithoutRole(ctx, platform, actor.Address())) + require.NoError(t, ledgerGiveBalance(ctx, platform, actor.Address(), initialUserFunds)) + // create_streams is 100 TRUF/stream (#3971) + 1 TRUF/insert (#3805), all + // charged against hoodi_tt. Fund 300 TRUF for headroom. + require.NoError(t, feefund.EnsureWalletFunded(ctx, platform, actor.Address(), "300000000000000000000"), + "fund actor on hoodi_tt for write fees") + + userLower := strings.ToLower(actor.Address()) + primitiveStream := util.GenerateStreamId("trim_e2e_primitive") + + // deployStream (method 1) at an old height — kept (wrong method). + createLeaderPub, _ := newLeader(t) + createTx, err := callActionWithLeader(ctx, platform, actor, createLeaderPub, 5, "create_streams", []any{ + []string{primitiveStream.String()}, + []string{"primitive"}, + }) + require.NoError(t, err) + + // insertRecords (method 2) at an old height — the trim target. + oldLeaderPub, _ := newLeader(t) + oldVal, err := kwilTypes.ParseDecimalExplicit("10.5", 36, 18) + require.NoError(t, err) + oldInsertTx, err := callActionWithLeader(ctx, platform, actor, oldLeaderPub, 6, "insert_records", []any{ + []string{userLower}, + []string{primitiveStream.String()}, + []int64{1000}, + []*kwilTypes.Decimal{oldVal}, + }) + require.NoError(t, err) + + // insertRecords (method 2) at a recent height — kept (>= cutoff). + newLeaderPub, _ := newLeader(t) + newVal, err := kwilTypes.ParseDecimalExplicit("11.5", 36, 18) + require.NoError(t, err) + newInsertTx, err := callActionWithLeader(ctx, platform, actor, newLeaderPub, 190_000, "insert_records", []any{ + []string{userLower}, + []string{primitiveStream.String()}, + []int64{2000}, + []*kwilTypes.Decimal{newVal}, + }) + require.NoError(t, err) + + // Sanity: the real write path recorded all three rows, and the old write + // carries its fee-distribution child. + requireTxEvent(t, ctx, platform, createTx, true) + requireTxEvent(t, ctx, platform, oldInsertTx, true) + requireTxEvent(t, ctx, platform, newInsertTx, true) + requireDistribution(t, ctx, platform, oldInsertTx, true) + + // Trim at height 200000 → cutoff = 200000 - 172800 = 27200. + require.NoError(t, callTrimTxEvents(ctx, platform, owner, 200_000, 172_800, 100)) + + requireTxEvent(t, ctx, platform, oldInsertTx, false) // old write pruned + requireDistribution(t, ctx, platform, oldInsertTx, false) // fee distribution cascaded away + requireTxEvent(t, ctx, platform, newInsertTx, true) // recent write kept + requireTxEvent(t, ctx, platform, createTx, true) // deployStream kept (wrong method) + + return nil + } +}