Use GRPC client, fix service cleanup

pkg/api/publishWorker.go

@@ -28,6 +28,7 @@ func StartPublishWorker(
 	reg *registrant.Registrant,
 	store *sql.DB,
 ) (*PublishWorker, error) {
+	log = log.Named("publishWorker")
 	q := queries.New(store)
 	query := func(ctx context.Context, lastSeenID int64, numRows int32) ([]queries.StagedOriginatorEnvelope, int64, error) {
 		results, err := q.SelectStagedOriginatorEnvelopes(

pkg/api/publish_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 func TestPublishEnvelope(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 
 	resp, err := svc.PublishEnvelope(
@@ -54,7 +54,7 @@ func TestPublishEnvelope(t *testing.T) {
 }
 
 func TestUnmarshalErrorOnPublish(t *testing.T) {
-	svc, _, cleanup := testutils.NewTestService(t)
+	svc, _, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 
 	envelope := testutils.CreatePayerEnvelope(t)
@@ -69,7 +69,7 @@ func TestUnmarshalErrorOnPublish(t *testing.T) {
 }
 
 func TestMismatchingOriginatorOnPublish(t *testing.T) {
-	svc, _, cleanup := testutils.NewTestService(t)
+	svc, _, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 
 	clientEnv := testutils.CreateClientEnvelope()
@@ -84,7 +84,7 @@ func TestMismatchingOriginatorOnPublish(t *testing.T) {
 }
 
 func TestMissingTopicOnPublish(t *testing.T) {
-	svc, _, cleanup := testutils.NewTestService(t)
+	svc, _, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 
 	clientEnv := testutils.CreateClientEnvelope()

pkg/api/query_test.go

@@ -64,7 +64,7 @@ func setupQueryTest(t *testing.T, db *sql.DB) []queries.InsertGatewayEnvelopePar
 }
 
 func TestQueryAllEnvelopes(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 	db_rows := setupQueryTest(t, db)
 
@@ -80,7 +80,7 @@ func TestQueryAllEnvelopes(t *testing.T) {
 }
 
 func TestQueryPagedEnvelopes(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 	db_rows := setupQueryTest(t, db)
 
@@ -96,7 +96,7 @@ func TestQueryPagedEnvelopes(t *testing.T) {
 }
 
 func TestQueryEnvelopesByOriginator(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 	db_rows := setupQueryTest(t, db)
 
@@ -117,7 +117,7 @@ func TestQueryEnvelopesByOriginator(t *testing.T) {
 }
 
 func TestQueryEnvelopesByTopic(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 	db_rows := setupQueryTest(t, db)
 
@@ -136,7 +136,7 @@ func TestQueryEnvelopesByTopic(t *testing.T) {
 }
 
 func TestQueryEnvelopesFromLastSeen(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 	db_rows := setupQueryTest(t, db)
 
@@ -155,7 +155,7 @@ func TestQueryEnvelopesFromLastSeen(t *testing.T) {
 }
 
 func TestQueryEnvelopesWithEmptyResult(t *testing.T) {
-	svc, db, cleanup := testutils.NewTestService(t)
+	svc, db, cleanup := testutils.NewTestAPIService(t)
 	defer cleanup()
 	db_rows := setupQueryTest(t, db)
 

pkg/api/server.go

@@ -106,6 +106,17 @@ func (s *ApiServer) Addr() net.Addr {
 	return s.grpcListener.Addr()
 }
 
+func (s *ApiServer) DialGRPC(ctx context.Context) (*grpc.ClientConn, error) {
+	// https://github.com/grpc/grpc/blob/master/doc/naming.md
+	dialAddr := fmt.Sprintf("passthrough://localhost/%s", s.grpcListener.Addr().String())
+	return grpc.DialContext(
+		ctx,
+		dialAddr,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultCallOptions(),
+	)
+}
+
 func (s *ApiServer) gracefulShutdown(timeout time.Duration) {
 	ctx, cancel := context.WithCancel(context.Background())
 	// Attempt to use GracefulStop up until the timeout

pkg/db/subscription.go

@@ -87,7 +87,7 @@ func (s *DBSubscription[ValueType, CursorType]) poll() {
 			break
 		} else if err != nil {
 			s.log.Error(
-				"Error querying for DB subscription",
+				fmt.Sprintf("Error querying for DB subscription: %v", err),
 				zap.Any("lastSeen", s.lastSeen),
 				zap.Int32("numRows", s.options.NumRows),
 			)

pkg/testutils/api_server.go

@@ -0,0 +1,81 @@
+package testutils
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/pingcap/log"
+	"github.com/stretchr/testify/require"
+	"github.com/xmtp/xmtpd/pkg/api"
+	"github.com/xmtp/xmtpd/pkg/db/queries"
+	mocks "github.com/xmtp/xmtpd/pkg/mocks/registry"
+	"github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api"
+	"github.com/xmtp/xmtpd/pkg/registrant"
+	"github.com/xmtp/xmtpd/pkg/registry"
+)
+
+func NewTestAPIServer(t *testing.T) (*api.ApiServer, *sql.DB, func()) {
+	ctx, cancel := context.WithCancel(context.Background())
+	log := NewLog(t)
+	db, _, dbCleanup := NewDB(t, ctx)
+	privKey, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	privKeyStr := "0x" + HexEncode(crypto.FromECDSA(privKey))
+	mockRegistry := mocks.NewMockNodeRegistry(t)
+	mockRegistry.EXPECT().GetNodes().Return([]registry.Node{
+		{NodeID: 1, SigningKey: &privKey.PublicKey},
+	}, nil)
+	registrant, err := registrant.NewRegistrant(ctx, queries.New(db), mockRegistry, privKeyStr)
+	require.NoError(t, err)
+
+	svr, err := api.NewAPIServer(ctx, db, log, 0 /*port*/, registrant, true /*enableReflection*/)
+	require.NoError(t, err)
+
+	return svr, db, func() {
+		log.Info("-------- Cleaning up server and DB ----------")
+		cancel()
+		svr.Close()
+		dbCleanup()
+	}
+}
+
+func NewTestAPIService(t *testing.T) (*api.Service, *sql.DB, func()) {
+	ctx, cancel := context.WithCancel(context.Background())
+	log := NewLog(t)
+	db, _, dbCleanup := NewDB(t, ctx)
+	privKey, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	privKeyStr := "0x" + HexEncode(crypto.FromECDSA(privKey))
+	mockRegistry := mocks.NewMockNodeRegistry(t)
+	mockRegistry.EXPECT().GetNodes().Return([]registry.Node{
+		{NodeID: 1, SigningKey: &privKey.PublicKey},
+	}, nil)
+	registrant, err := registrant.NewRegistrant(ctx, queries.New(db), mockRegistry, privKeyStr)
+	require.NoError(t, err)
+
+	svc, err := api.NewReplicationApiService(ctx, log, registrant, db)
+	require.NoError(t, err)
+
+	return svc, db, func() {
+		log.Info("-------- Cleaning up service and DB ----------")
+		cancel()
+		svc.Close()
+		dbCleanup()
+	}
+}
+
+func NewTestAPIClient(t *testing.T) (message_api.ReplicationApiClient, *sql.DB, func()) {
+	svc, db, cleanup := NewTestAPIServer(t)
+	conn, err := svc.DialGRPC(context.Background())
+	require.NoError(t, err)
+	client := message_api.NewReplicationApiClient(conn)
+
+	return client, db, func() {
+		log.Info("-------- Closing client connection ----------")
+		conn.Close()
+		require.NoError(t, err)
+		cleanup()
+	}
+}