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

Author: Aman035

universalClient/chains/common/event_cleaner_race_test.go

@@ -3,6 +3,7 @@ package common
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -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_cleaner_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")
+}