test(jsonrpc): avoid different test package setup to interfer with one another

- Added a PostgreSQL advisory lock held for the full test-process lifetime.
- Applied it to all three schema-resetting packages:
    - internal/jsonrpc
    - internal/repository/postgres
    - test/validator

- JSON-RPC services now clone the handler dispatch table.
- Test handlers modify only their service instance, not the package-global map.
- Added a regression test proving handler overrides do not leak across services.

Author: renatomaia

internal/jsonrpc/batchbudget_test.go

@@ -103,7 +103,7 @@ func TestBatchListItemLimitNormalizesLimitsLikeHandlers(t *testing.T) {
 func TestJSONRPCBatchRejectsListWorkOverLimitBeforeDispatch(t *testing.T) {
 	s := newBatchTestService()
 	var calls atomic.Int32
-	withTestRPCHandler(t, "cartesi_listApplications", func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, "cartesi_listApplications", func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		return true, nil
 	})
@@ -123,7 +123,7 @@ func TestJSONRPCBatchRejectsListWorkOverLimitBeforeDispatch(t *testing.T) {
 func TestJSONRPCBatchAllowsListWorkAtLimit(t *testing.T) {
 	s := newBatchTestService()
 	var calls atomic.Int32
-	withTestRPCHandler(t, "cartesi_listApplications", func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, "cartesi_listApplications", func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		return true, nil
 	})

internal/jsonrpc/batchcalls_test.go

@@ -41,6 +41,7 @@ func newBatchTestService() *Service {
 		Service: service.Service{
 			Logger: slog.New(slog.NewTextHandler(&bytes.Buffer{}, nil)),
 		},
+		handlers: cloneDispatchTable(jsonrpcHandlers),
 	}
 }
 
@@ -128,7 +129,7 @@ func TestJSONRPCBatchRejectsMoreThanMaximumBeforeDispatch(t *testing.T) {
 	s := newBatchTestService()
 	var calls atomic.Int32
 	const method = "test_batch_cap"
-	withTestRPCHandler(t, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		return true, nil
 	})
@@ -312,7 +313,7 @@ func TestJSONRPCBatchReplacesResponsesAtCumulativeResponseBudget(t *testing.T) {
 	var calls atomic.Int32
 	const method = "test_large_batch_result"
 	largeResult := strings.Repeat("x", testLargeResultSize)
-	withTestRPCHandler(t, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		return largeResult, nil
 	})
@@ -351,7 +352,7 @@ func TestJSONRPCBatchStopsBetweenEntriesWhenContextIsCanceled(t *testing.T) {
 	s.Logger = slog.New(slog.NewJSONHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
 	ctx, cancel := context.WithCancel(context.Background())
 	const method = "test_cancel_batch"
-	withTestRPCHandler(t, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		cancel()
 		return true, nil
@@ -387,7 +388,7 @@ func TestJSONRPCBatchStopsSilentlyWhenRepositoryCallIsCanceled(t *testing.T) {
 	s.Logger = slog.New(slog.NewJSONHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
 	ctx, cancel := context.WithCancel(context.Background())
 	const method = "test_repository_cancel_batch"
-	withTestRPCHandler(t, method, func(s *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(s *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		cancel()
 		return nil, s.repositoryError("Unable to retrieve test data from repository",
@@ -419,7 +420,7 @@ func TestJSONRPCBatchReturnsErrorsForIDDRequestsAfterDeadline(t *testing.T) {
 	s := newBatchTestService()
 	var calls atomic.Int32
 	const method = "test_deadline_batch"
-	withTestRPCHandler(t, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		calls.Add(1)
 		return true, nil
 	})
@@ -462,10 +463,10 @@ func TestJSONRPCBatchRecoversPanicPerEntry(t *testing.T) {
 		panicMethod = "test_panic_batch"
 		okMethod    = "test_after_panic_batch"
 	)
-	withTestRPCHandler(t, panicMethod, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, panicMethod, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		panic("test panic")
 	})
-	withTestRPCHandler(t, okMethod, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, okMethod, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		return "ok", nil
 	})
 
@@ -496,7 +497,7 @@ func TestJSONRPCBatchUsesOneAdmissionPermit(t *testing.T) {
 	}
 	var nestedAcquisitions atomic.Int32
 	const method = "test_batch_admission"
-	withTestRPCHandler(t, method, func(s *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(s *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		if s.admission.TryAcquire() {
 			nestedAcquisitions.Add(1)
 			s.admission.Release()
@@ -578,15 +579,31 @@ func TestJSONRPCBatchMethodLoggingIsTruncated(t *testing.T) {
 	require.True(t, found)
 }
 
-func withTestRPCHandler(t *testing.T, method string, handler rpcHandler) {
+func withTestRPCHandler(t *testing.T, service *Service, method string, handler rpcHandler) {
 	t.Helper()
-	previous, existed := jsonrpcHandlers[method]
-	jsonrpcHandlers[method] = handler
+	previous, existed := service.handlers[method]
+	service.handlers[method] = handler
 	t.Cleanup(func() {
 		if existed {
-			jsonrpcHandlers[method] = previous
+			service.handlers[method] = previous
 		} else {
-			delete(jsonrpcHandlers, method)
+			delete(service.handlers, method)
 		}
 	})
 }
+
+func TestRPCHandlerOverridesAreServiceLocal(t *testing.T) {
+	first := newBatchTestService()
+	second := newBatchTestService()
+	const method = "test_service_local_handler"
+	withTestRPCHandler(t, first, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+		return true, nil
+	})
+
+	_, firstHasHandler := first.handlers[method]
+	_, secondHasHandler := second.handlers[method]
+	_, globalHasHandler := jsonrpcHandlers[method]
+	require.True(t, firstHasHandler)
+	require.False(t, secondHasHandler)
+	require.False(t, globalHasHandler)
+}

internal/jsonrpc/jsonrpc.go

@@ -74,6 +74,14 @@ const (
 type rpcHandler = func(*Service, *http.Request, RPCRequest) (any, error)
 type dispatchTable = map[string]rpcHandler
 
+func cloneDispatchTable(source dispatchTable) dispatchTable {
+	clone := make(dispatchTable, len(source))
+	for method, handler := range source {
+		clone[method] = handler
+	}
+	return clone
+}
+
 var jsonrpcHandlers = dispatchTable{
 	"rpc.discover":                            handleDiscover,
 	"cartesi_listApplications":                handleListApplications,
@@ -212,7 +220,7 @@ func (s *Service) handleRequest(w io.Writer, r *http.Request, req RPCRequest) er
 	if req.JSONRPC != "2.0" || req.Method == "" {
 		return writeRPCError(w, req.ID, JSONRPC_INVALID_REQUEST, "invalid request")
 	}
-	fn, ok := jsonrpcHandlers[req.Method]
+	fn, ok := s.handlers[req.Method]
 	if !ok {
 		s.Logger.Debug("RPC method not found", "method", truncatedMethod(req.Method))
 		return writeRPCError(w, req.ID, JSONRPC_METHOD_NOT_FOUND, "Method not found")

internal/jsonrpc/jsonrpc_test.go

@@ -86,7 +86,7 @@ func TestJSONRPCSingleRequestReplacesResponseAtResponseBudget(t *testing.T) {
 	const method = "test_large_single_result"
 	largeResult := strings.Repeat("x", MAX_RESPONSE_SIZE)
 	var called bool
-	withTestRPCHandler(t, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
 		called = true
 		return largeResult, nil
 	})

internal/jsonrpc/main_test.go

@@ -0,0 +1,28 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package jsonrpc
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/cartesi/rollups-node/test/tooling/db"
+)
+
+func TestMain(m *testing.M) {
+	endpoint, err := db.GetTestDatabaseEndpoint()
+	if err != nil {
+		os.Exit(m.Run())
+	}
+	release, err := db.LockTestPostgres(context.Background(), endpoint)
+	if err != nil {
+		_, _ = fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	code := m.Run()
+	release()
+	os.Exit(code)
+}

internal/jsonrpc/service.go

@@ -43,6 +43,7 @@ type Service struct {
 	listen func(network, address string) (net.Listener, error)
 	// OpenAPI description for JSON-RPC API loaded from 'jsonrpc-discover.json' file
 	discoverSpec any
+	handlers     dispatchTable
 	// dispatchTimeout expires requests early enough to serialize a complete
 	// timeout response before the HTTP server's write deadline.
 	dispatchTimeout time.Duration
@@ -71,6 +72,7 @@ func Create(ctx context.Context, c *CreateInfo) (*Service, error) {
 	}
 
 	s.repository = c.Repository
+	s.handlers = cloneDispatchTable(jsonrpcHandlers)
 	if s.repository == nil {
 		return nil, fmt.Errorf("repository on validator service Create is nil")
 	}

internal/jsonrpc/service_test.go

@@ -59,7 +59,7 @@ func TestJSONRPC_ServerHandlerAppliesBatchDispatchTimeout(t *testing.T) {
 	s.dispatchTimeout = 10 * time.Millisecond
 
 	const method = "test_server_dispatch_timeout"
-	withTestRPCHandler(t, method, func(_ *Service, r *http.Request, _ RPCRequest) (any, error) {
+	withTestRPCHandler(t, s, method, func(_ *Service, r *http.Request, _ RPCRequest) (any, error) {
 		<-r.Context().Done()
 		return true, nil
 	})

internal/repository/postgres/postgres_repo_test.go

@@ -5,6 +5,8 @@ package postgres_test
 
 import (
 	"context"
+	"fmt"
+	"os"
 	"testing"
 
 	"github.com/cartesi/rollups-node/internal/repository"
@@ -14,6 +16,21 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestMain(m *testing.M) {
+	endpoint, err := db.GetTestDatabaseEndpoint()
+	if err != nil {
+		os.Exit(m.Run())
+	}
+	release, err := db.LockTestPostgres(context.Background(), endpoint)
+	if err != nil {
+		_, _ = fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	code := m.Run()
+	release()
+	os.Exit(code)
+}
+
 func TestPostgresRepository(t *testing.T) {
 	endpoint, err := db.GetTestDatabaseEndpoint()
 	if err != nil {

test/tooling/db/db.go

@@ -4,12 +4,16 @@
 package db
 
 import (
+	"context"
 	"fmt"
 	"os"
 
 	"github.com/cartesi/rollups-node/internal/repository/postgres/schema"
+	"github.com/jackc/pgx/v5"
 )
 
+const testDatabaseLockID int64 = 0x4352545349544553 // "CRTSITES"
+
 func GetTestDatabaseEndpoint() (string, error) {
 	endpoint, ok := os.LookupEnv("CARTESI_TEST_DATABASE_CONNECTION")
 	if !ok {
@@ -18,6 +22,21 @@ func GetTestDatabaseEndpoint() (string, error) {
 	return endpoint, nil
 }
 
+// LockTestPostgres serializes package test processes that reset the shared test
+// schema. The session-level advisory lock is held until the returned connection
+// closer is called.
+func LockTestPostgres(ctx context.Context, endpoint string) (func(), error) {
+	conn, err := pgx.Connect(ctx, endpoint)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect for test database lock: %w", err)
+	}
+	if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", testDatabaseLockID); err != nil {
+		_ = conn.Close(context.Background())
+		return nil, fmt.Errorf("failed to lock test database: %w", err)
+	}
+	return func() { _ = conn.Close(context.Background()) }, nil
+}
+
 func SetupTestPostgres(endpoint string) error {
 
 	schema, err := schema.New(endpoint)

test/validator/validator_test.go

@@ -5,8 +5,10 @@ package validator
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"math/big"
+	"os"
 	"testing"
 	"time"
 
@@ -29,6 +31,21 @@ const MAX_OUTPUT_TREE_HEIGHT = merkle.TREE_DEPTH //nolint: revive
 
 const testTimeout = 300 * time.Second
 
+func TestMain(m *testing.M) {
+	endpoint, err := db.GetTestDatabaseEndpoint()
+	if err != nil {
+		os.Exit(m.Run())
+	}
+	release, err := db.LockTestPostgres(context.Background(), endpoint)
+	if err != nil {
+		_, _ = fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	code := m.Run()
+	release()
+	os.Exit(code)
+}
+
 type ValidatorRepositoryIntegrationSuite struct {
 	suite.Suite
 	ctx              context.Context