fix: F-2026-18800 | [Dual Defense] Short SVM send_funds Defaults TxType=0 While Claiming Funds → Mapped to GAS (#327)

* fix(chains): reject truncated UniversalTx events instead of defaulting tx_type to GAS

* fix(common): guard event cleaner lifecycle with a mutex and wait for its goroutine

* revert unrelated gofmt change in client_test

* test: fold new tests into the existing per-source test files

Author: Aman035

universalClient/chains/common/event_cleaner.go

@@ -3,6 +3,7 @@ package common
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/pushchain/push-chain-node/universalClient/db"
@@ -23,9 +24,14 @@ type EventCleaner struct {
 	cleanupInterval time.Duration
 	retentionPeriod time.Duration
 	logger          zerolog.Logger
-	ticker          *time.Ticker
-	stopCh          chan struct{}
-	running         bool
+
+	// mu guards running and stopCh, which Start and Stop both touch. The
+	// cleanup goroutine reads neither: it closes over its own copies, so the
+	// only cross-goroutine state is the channel it selects on.
+	mu      sync.Mutex
+	running bool
+	stopCh  chan struct{}
+	wg      sync.WaitGroup
 }
 
 // NewEventCleaner creates a new event cleaner for a chain
@@ -54,9 +60,16 @@ func NewEventCleaner(
 
 // Start begins the periodic cleanup process
 func (ec *EventCleaner) Start(ctx context.Context) error {
+	ec.mu.Lock()
 	if ec.running {
+		ec.mu.Unlock()
 		return fmt.Errorf("event cleaner is already running")
 	}
+	stopCh := make(chan struct{})
+	ec.running = true
+	ec.stopCh = stopCh
+	ec.wg.Add(1)
+	ec.mu.Unlock()
 
 	ec.logger.Debug().
 		Str("cleanup_interval", ec.cleanupInterval.String()).
@@ -69,21 +82,23 @@ func (ec *EventCleaner) Start(ctx context.Context) error {
 		// Don't fail startup on cleanup error, just log it
 	}
 
-	ec.running = true
-	ec.stopCh = make(chan struct{})
-	ec.ticker = time.NewTicker(ec.cleanupInterval)
+	// The ticker and stop channel are the goroutine's own. Holding them on the
+	// struct let Stop write the fields while the goroutine was still reading
+	// them, which is the race this shape removes.
+	ticker := time.NewTicker(ec.cleanupInterval)
 
 	go func() {
-		defer ec.ticker.Stop()
+		defer ec.wg.Done()
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ctx.Done():
 				ec.logger.Debug().Msg("context cancelled, stopping event cleaner")
 				return
-			case <-ec.stopCh:
+			case <-stopCh:
 				ec.logger.Debug().Msg("stop signal received, stopping event cleaner")
 				return
-			case <-ec.ticker.C:
+			case <-ticker.C:
 				if err := ec.performCleanup(); err != nil {
 					ec.logger.Error().Err(err).Msg("failed to perform scheduled cleanup")
 				}
@@ -94,17 +109,24 @@ func (ec *EventCleaner) Start(ctx context.Context) error {
 	return nil
 }
 
-// Stop gracefully stops the event cleaner. No-op if not running.
+// Stop gracefully stops the event cleaner and waits for the cleanup goroutine
+// to exit. No-op if not running.
+//
+// Waiting matters on shutdown: the goroutine runs queries against the chain
+// database, and returning before it finishes lets the caller close that
+// database underneath an in-flight cleanup.
 func (ec *EventCleaner) Stop() {
+	ec.mu.Lock()
 	if !ec.running {
+		ec.mu.Unlock()
 		return
 	}
 	ec.logger.Debug().Msg("stopping event cleaner")
-	if ec.ticker != nil {
-		ec.ticker.Stop()
-	}
-	close(ec.stopCh)
 	ec.running = false
+	close(ec.stopCh)
+	ec.mu.Unlock()
+
+	ec.wg.Wait()
 }
 
 // performCleanup executes cleanup of terminal events (COMPLETED, REORGED, REVERTED)

universalClient/chains/common/event_cleaner_test.go

@@ -3,6 +3,7 @@ package common
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -75,8 +76,8 @@ func TestEventCleanerStruct(t *testing.T) {
 		assert.Nil(t, ec.database)
 		assert.Equal(t, time.Duration(0), ec.cleanupInterval)
 		assert.Equal(t, time.Duration(0), ec.retentionPeriod)
-		assert.Nil(t, ec.ticker)
 		assert.Nil(t, ec.stopCh)
+		assert.False(t, ec.running)
 	})
 }
 
@@ -250,7 +251,7 @@ func TestEventCleanerStart(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 
 		require.NoError(t, cleaner.Start(ctx))
-		require.NotNil(t, cleaner.ticker)
+		require.NotNil(t, cleaner.stopCh)
 
 		cancel()
 		time.Sleep(100 * time.Millisecond)
@@ -337,3 +338,138 @@ func TestEventCleanerStartStopLifecycle(t *testing.T) {
 		time.Sleep(50 * time.Millisecond)
 	})
 }
+
+// Start and Stop race against the cleanup goroutine. Run under -race.
+func TestEventCleaner_StartStopUnderRace(t *testing.T) {
+	for i := 0; i < 20; i++ {
+		database := newTestCleanerDB(t, nil)
+		cleaner := NewEventCleaner(database, intPtr(3600), intPtr(0), "test-chain", zerolog.Nop())
+		// Fast enough that the goroutine is inside performCleanup while Stop runs.
+		cleaner.cleanupInterval = time.Millisecond
+
+		require.NoError(t, cleaner.Start(context.Background()))
+		cleaner.Stop()
+	}
+}
+
+// Concurrent Stop calls must not double close the channel or return before the
+// goroutine has exited.
+func TestEventCleaner_ConcurrentStop(t *testing.T) {
+	database := newTestCleanerDB(t, nil)
+	cleaner := NewEventCleaner(database, intPtr(3600), intPtr(0), "test-chain", zerolog.Nop())
+	cleaner.cleanupInterval = time.Millisecond
+
+	require.NoError(t, cleaner.Start(context.Background()))
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			cleaner.Stop()
+		}()
+	}
+	wg.Wait()
+
+	assert.False(t, cleaner.running)
+}
+
+// Concurrent Start calls must leave exactly one goroutine running.
+func TestEventCleaner_ConcurrentStart(t *testing.T) {
+	database := newTestCleanerDB(t, nil)
+	cleaner := NewEventCleaner(database, intPtr(3600), intPtr(0), "test-chain", zerolog.Nop())
+	cleaner.cleanupInterval = time.Millisecond
+
+	var mu sync.Mutex
+	started := 0
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := cleaner.Start(context.Background()); err == nil {
+				mu.Lock()
+				started++
+				mu.Unlock()
+			}
+		}()
+	}
+	wg.Wait()
+
+	assert.Equal(t, 1, started, "more than one cleanup goroutine was started")
+	cleaner.Stop()
+}
+
+// Stop must not return while a cleanup is still in flight, otherwise the caller
+// can close the chain database underneath an in-flight query.
+//
+// Held open with a write transaction so the goroutine is genuinely blocked
+// inside performCleanup while Stop is called. Without that, the goroutine exits
+// so fast that a Stop which does not wait looks identical to one that does.
+func TestEventCleaner_StopWaitsForInFlightCleanup(t *testing.T) {
+	database := newTestCleanerDB(t, nil)
+	cleaner := NewEventCleaner(database, intPtr(3600), intPtr(0), "test-chain", zerolog.Nop())
+	cleaner.cleanupInterval = time.Millisecond
+
+	// Start first: the initial cleanup is synchronous and would block on the lock.
+	require.NoError(t, cleaner.Start(context.Background()))
+
+	// Take the write lock so the next ticked cleanup blocks on DELETE.
+	tx := database.Client().Begin()
+	require.NoError(t, tx.Error)
+	require.NoError(t, tx.Exec(
+		"CREATE TABLE IF NOT EXISTS lock_probe (id INTEGER PRIMARY KEY)").Error)
+	require.NoError(t, tx.Exec("INSERT INTO lock_probe (id) VALUES (1)").Error)
+
+	time.Sleep(50 * time.Millisecond) // let a tick land and block
+
+	stopped := make(chan struct{})
+	go func() {
+		cleaner.Stop()
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+		tx.Rollback()
+		t.Fatal("Stop returned while a cleanup was still in flight")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	tx.Rollback() // release the lock; the cleanup can now finish
+
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Stop did not return after the cleanup finished")
+	}
+}
+
+// Cancelling the context stops the goroutine, and a later Stop is still safe.
+func TestEventCleaner_ContextCancelThenStop(t *testing.T) {
+	database := newTestCleanerDB(t, nil)
+	cleaner := NewEventCleaner(database, intPtr(3600), intPtr(0), "test-chain", zerolog.Nop())
+	cleaner.cleanupInterval = time.Millisecond
+
+	ctx, cancel := context.WithCancel(context.Background())
+	require.NoError(t, cleaner.Start(ctx))
+	cancel()
+	time.Sleep(20 * time.Millisecond)
+
+	cleaner.Stop() // must not hang or panic
+	assert.False(t, cleaner.running)
+}
+
+// Restart after Stop gets a fresh channel rather than reusing the closed one.
+func TestEventCleaner_RestartAfterStop(t *testing.T) {
+	database := newTestCleanerDB(t, nil)
+	cleaner := NewEventCleaner(database, intPtr(3600), intPtr(0), "test-chain", zerolog.Nop())
+	cleaner.cleanupInterval = time.Millisecond
+
+	require.NoError(t, cleaner.Start(context.Background()))
+	cleaner.Stop()
+
+	require.NoError(t, cleaner.Start(context.Background()), "restart was refused")
+	cleaner.Stop()
+}

universalClient/chains/common/event_processor_test.go

@@ -1054,3 +1054,77 @@ func TestProcessConfirmedEventsEnabledFlags(t *testing.T) {
 		assert.Equal(t, store.StatusConfirmed, inboundEvt.Status)
 	})
 }
+
+// The wire values the gateways emit are 0-indexed (Gas, GasAndPayload, Funds,
+// FundsAndPayload) while the chain enum reserves 0 for UNSPECIFIED, so the
+// mapping is shifted by one. A decoder that leaves TxType unset therefore does
+// not produce "unknown", it produces GAS.
+func TestConstructInbound_TxTypeMapping(t *testing.T) {
+	processor := &EventProcessor{}
+
+	for _, tc := range []struct {
+		wire uint
+		want uexecutortypes.TxType
+	}{
+		{0, uexecutortypes.TxType_GAS},
+		{1, uexecutortypes.TxType_GAS_AND_PAYLOAD},
+		{2, uexecutortypes.TxType_FUNDS},
+		{3, uexecutortypes.TxType_FUNDS_AND_PAYLOAD},
+		{4, uexecutortypes.TxType_UNSPECIFIED_TX},
+		{99, uexecutortypes.TxType_UNSPECIFIED_TX},
+	} {
+		data, err := json.Marshal(UniversalTx{
+			SourceChain: "solana:devnet",
+			Sender:      "0xabc",
+			Recipient:   "0xdef",
+			Amount:      "5000000",
+			TxType:      tc.wire,
+		})
+		require.NoError(t, err)
+
+		inbound, err := processor.constructInbound(&store.Event{
+			EventID:   "sig:0",
+			EventData: data,
+		})
+		require.NoError(t, err)
+		assert.Equal(t, tc.want, inbound.TxType, "wire value %d", tc.wire)
+	}
+}
+
+// A FUNDS transfer must never reach the keeper as GAS. The two dispatch to
+// different handlers: GAS mints and autoswaps into the sender UEA, FUNDS
+// deposits PRC20 to the recipient, so the same amount lands with a different
+// party. This is the end to end assertion the finding asks for.
+func TestConstructInbound_FundsNeverBecomesGas(t *testing.T) {
+	processor := &EventProcessor{}
+
+	data, err := json.Marshal(UniversalTx{
+		SourceChain: "solana:devnet",
+		Sender:      "0xabc",
+		Recipient:   "0xdef",
+		Amount:      "5000000",
+		TxType:      2, // Funds, as the real devnet events carry
+	})
+	require.NoError(t, err)
+
+	inbound, err := processor.constructInbound(&store.Event{
+		EventID:   "sig:0",
+		EventData: data,
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, uexecutortypes.TxType_FUNDS, inbound.TxType)
+	assert.NotEqual(t, uexecutortypes.TxType_GAS, inbound.TxType,
+		"a FUNDS transfer routed to GAS credits the sender instead of the recipient")
+}
+
+// An event whose data never made it past the decoder must be refused outright
+// rather than defaulted. The parsers now discard such events, so this is the
+// backstop if one ever reaches the store.
+func TestConstructInbound_RejectsEventWithoutData(t *testing.T) {
+	processor := &EventProcessor{}
+
+	_, err := processor.constructInbound(&store.Event{EventID: "sig:0"})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "event data is missing")
+}