test(e2e): add indexing and gRPC query e2e tests

.github/workflows/ci.yml

@@ -42,8 +42,8 @@ jobs:
           go-version-file: go.mod
       - run: go build -o bin/apex ./cmd/apex
 
-  e2e-submission:
-    name: E2E Submission
+  e2e:
+    name: E2E
     runs-on: ubuntu-latest
     timeout-minutes: 30
     steps:
@@ -55,4 +55,4 @@ jobs:
             go.sum
             e2e/go.sum
       - working-directory: e2e
-        run: go test -race -count=1 -timeout 20m -run TestSubmissionViaJSONRPC ./...
+        run: go test -race -count=1 -timeout 20m ./...

e2e/grpc_test.go

@@ -0,0 +1,178 @@
+package e2e
+
+import (
+	"bytes"
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	pb "github.com/evstack/apex/pkg/api/grpc/gen/apex/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func TestGRPCBlobQuery(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping Docker-backed e2e test in short mode")
+	}
+
+	setupCtx, cancelSetup := context.WithTimeout(context.Background(), chainStartupTimeout)
+	defer cancelSetup()
+
+	newRPCCtx := func(t *testing.T) (context.Context, context.CancelFunc) {
+		t.Helper()
+		return context.WithTimeout(context.Background(), 10*time.Second)
+	}
+
+	grpcAddr, chainID, signerKeyHex, signerAddress := startSubmissionTestChain(t, setupCtx)
+	namespace := testNamespace(t, []byte("apex-grpc"))
+	data := []byte("apex grpc query e2e")
+	commitment := mustBlobCommitment(t, namespace, data)
+
+	apexBinary := buildApexBinary(t)
+	apexRPCAddr := reserveTCPAddr(t)
+	apexGRPCAddr := reserveTCPAddr(t)
+	keyPath := writeSignerKey(t, signerKeyHex)
+	configPath := writeApexConfig(t, apexConfig{
+		Namespace:       namespace,
+		DataGRPCAddr:    grpcAddr,
+		SubmissionGRPC:  grpcAddr,
+		ChainID:         chainID,
+		SignerKeyPath:   keyPath,
+		StoragePath:     filepath.Join(t.TempDir(), "apex.db"),
+		RPCListenAddr:   apexRPCAddr,
+		GRPCListenAddr:  apexGRPCAddr,
+		GasPrice:        submissionGasPrice,
+		MaxGasPrice:     submissionGasPrice,
+		ConfirmTimeoutS: submissionConfirmTimeout,
+	})
+
+	proc := startApexProcess(t, apexBinary, configPath)
+	defer proc.Stop(t)
+	waitForApexHTTP(t, proc, apexRPCAddr)
+
+	resp := doRPC(t, proc, apexRPCAddr, "blob.Submit",
+		[]map[string]any{{
+			"namespace":     namespace,
+			"data":          data,
+			"share_version": 0,
+			"commitment":    commitment,
+			"index":         -1,
+		}},
+		map[string]any{
+			"gas_price":        submissionGasPrice,
+			"is_gas_price_set": true,
+		},
+	)
+	if resp.Error != nil {
+		t.Fatalf("blob.Submit error: %s", resp.Error.Message)
+	}
+	height := decodeSubmitHeight(t, resp.Result)
+	if height == 0 {
+		t.Fatal("submission height must be positive")
+	}
+
+	waitForIndexedBlob(t, proc, apexRPCAddr, commitment, data, namespace, signerAddress)
+
+	conn, err := grpc.NewClient(apexGRPCAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("dial apex gRPC: %v", err)
+	}
+	defer conn.Close() //nolint:errcheck
+
+	blobClient := pb.NewBlobServiceClient(conn)
+	headerClient := pb.NewHeaderServiceClient(conn)
+
+	t.Run("BlobService.GetByCommitment", func(t *testing.T) {
+		rpcCtx, cancel := newRPCCtx(t)
+		defer cancel()
+		r, err := blobClient.GetByCommitment(rpcCtx, &pb.GetByCommitmentRequest{Commitment: commitment})
+		if err != nil {
+			t.Fatalf("GetByCommitment: %v", err)
+		}
+		if r.GetBlob() == nil {
+			t.Fatal("expected non-nil blob")
+		}
+		if !bytes.Equal(r.GetBlob().GetData(), data) {
+			t.Fatalf("data mismatch: got %q want %q", r.GetBlob().GetData(), data)
+		}
+		if !bytes.Equal(r.GetBlob().GetCommitment(), commitment) {
+			t.Fatalf("commitment mismatch: got %x want %x", r.GetBlob().GetCommitment(), commitment)
+		}
+	})
+
+	t.Run("BlobService.Get", func(t *testing.T) {
+		rpcCtx, cancel := newRPCCtx(t)
+		defer cancel()
+		r, err := blobClient.Get(rpcCtx, &pb.GetRequest{
+			Height:     height,
+			Namespace:  namespace,
+			Commitment: commitment,
+		})
+		if err != nil {
+			t.Fatalf("Get: %v", err)
+		}
+		if !bytes.Equal(r.GetBlob().GetData(), data) {
+			t.Fatalf("data mismatch: got %q want %q", r.GetBlob().GetData(), data)
+		}
+		if !bytes.Equal(r.GetBlob().GetCommitment(), commitment) {
+			t.Fatalf("commitment mismatch: got %x want %x", r.GetBlob().GetCommitment(), commitment)
+		}
+	})
+
+	t.Run("BlobService.GetAll", func(t *testing.T) {
+		rpcCtx, cancel := newRPCCtx(t)
+		defer cancel()
+		r, err := blobClient.GetAll(rpcCtx, &pb.GetAllRequest{
+			Height:     height,
+			Namespaces: [][]byte{namespace},
+		})
+		if err != nil {
+			t.Fatalf("GetAll: %v", err)
+		}
+		if len(r.GetBlobs()) == 0 {
+			t.Fatal("expected at least one blob")
+		}
+		found := false
+		for _, b := range r.GetBlobs() {
+			if bytes.Equal(b.GetCommitment(), commitment) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Fatalf("blob with commitment %x not in GetAll result", commitment)
+		}
+	})
+
+	t.Run("HeaderService.GetByHeight", func(t *testing.T) {
+		rpcCtx, cancel := newRPCCtx(t)
+		defer cancel()
+		r, err := headerClient.GetByHeight(rpcCtx, &pb.GetByHeightRequest{Height: height})
+		if err != nil {
+			t.Fatalf("GetByHeight: %v", err)
+		}
+		if r.GetHeader() == nil {
+			t.Fatal("expected non-nil header")
+		}
+		if r.GetHeader().GetHeight() != height {
+			t.Fatalf("header height = %d, want %d", r.GetHeader().GetHeight(), height)
+		}
+	})
+
+	t.Run("HeaderService.LocalHead", func(t *testing.T) {
+		rpcCtx, cancel := newRPCCtx(t)
+		defer cancel()
+		r, err := headerClient.LocalHead(rpcCtx, &pb.LocalHeadRequest{})
+		if err != nil {
+			t.Fatalf("LocalHead: %v", err)
+		}
+		if r.GetHeader() == nil {
+			t.Fatal("expected non-nil header")
+		}
+		if r.GetHeader().GetHeight() < height {
+			t.Fatalf("local head height %d < submission height %d", r.GetHeader().GetHeight(), height)
+		}
+	})
+}

e2e/indexing_test.go

@@ -0,0 +1,129 @@
+package e2e
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"path/filepath"
+	"testing"
+)
+
+func TestIndexingQueryPaths(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping Docker-backed e2e test in short mode")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), chainStartupTimeout)
+	defer cancel()
+
+	grpcAddr, chainID, signerKeyHex, signerAddress := startSubmissionTestChain(t, ctx)
+	namespace := testNamespace(t, []byte("apex-idx"))
+	data := []byte("apex indexing query e2e")
+	commitment := mustBlobCommitment(t, namespace, data)
+
+	apexBinary := buildApexBinary(t)
+	apexRPCAddr := reserveTCPAddr(t)
+	apexGRPCAddr := reserveTCPAddr(t)
+	keyPath := writeSignerKey(t, signerKeyHex)
+	configPath := writeApexConfig(t, apexConfig{
+		Namespace:       namespace,
+		DataGRPCAddr:    grpcAddr,
+		SubmissionGRPC:  grpcAddr,
+		ChainID:         chainID,
+		SignerKeyPath:   keyPath,
+		StoragePath:     filepath.Join(t.TempDir(), "apex.db"),
+		RPCListenAddr:   apexRPCAddr,
+		GRPCListenAddr:  apexGRPCAddr,
+		GasPrice:        submissionGasPrice,
+		MaxGasPrice:     submissionGasPrice,
+		ConfirmTimeoutS: submissionConfirmTimeout,
+	})
+
+	proc := startApexProcess(t, apexBinary, configPath)
+	defer proc.Stop(t)
+	waitForApexHTTP(t, proc, apexRPCAddr)
+
+	resp := doRPC(t, proc, apexRPCAddr, "blob.Submit",
+		[]map[string]any{{
+			"namespace":     namespace,
+			"data":          data,
+			"share_version": 0,
+			"commitment":    commitment,
+			"index":         -1,
+		}},
+		map[string]any{
+			"gas_price":        submissionGasPrice,
+			"is_gas_price_set": true,
+		},
+	)
+	if resp.Error != nil {
+		t.Fatalf("blob.Submit error: %s", resp.Error.Message)
+	}
+	height := decodeSubmitHeight(t, resp.Result)
+	if height == 0 {
+		t.Fatal("submission height must be positive")
+	}
+
+	waitForIndexedBlob(t, proc, apexRPCAddr, commitment, data, namespace, signerAddress)
+
+	t.Run("blob.Get", func(t *testing.T) {
+		r := doRPC(t, proc, apexRPCAddr, "blob.Get", height, namespace, commitment)
+		if r.Error != nil {
+			t.Fatalf("blob.Get error: %s", r.Error.Message)
+		}
+		var b rpcBlob
+		if err := json.Unmarshal(r.Result, &b); err != nil {
+			t.Fatalf("decode blob: %v", err)
+		}
+		if !bytes.Equal(b.Data, data) {
+			t.Fatalf("data mismatch: got %q want %q", b.Data, data)
+		}
+		if !bytes.Equal(b.Commitment, commitment) {
+			t.Fatalf("commitment mismatch: got %x want %x", b.Commitment, commitment)
+		}
+	})
+
+	t.Run("blob.GetAll", func(t *testing.T) {
+		r := doRPC(t, proc, apexRPCAddr, "blob.GetAll", height, [][]byte{namespace})
+		if r.Error != nil {
+			t.Fatalf("blob.GetAll error: %s", r.Error.Message)
+		}
+		var blobs []rpcBlob
+		if err := json.Unmarshal(r.Result, &blobs); err != nil {
+			t.Fatalf("decode blobs: %v", err)
+		}
+		if len(blobs) == 0 {
+			t.Fatal("expected at least one blob from GetAll")
+		}
+		found := false
+		for _, b := range blobs {
+			if bytes.Equal(b.Commitment, commitment) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Fatalf("blob with commitment %x not found in GetAll result", commitment)
+		}
+	})
+
+	t.Run("header.GetByHeight", func(t *testing.T) {
+		r := doRPC(t, proc, apexRPCAddr, "header.GetByHeight", height)
+		if r.Error != nil {
+			t.Fatalf("header.GetByHeight error: %s", r.Error.Message)
+		}
+		if len(r.Result) == 0 || string(r.Result) == "null" {
+			t.Fatal("expected non-null header")
+		}
+	})
+
+	t.Run("header.LocalHead", func(t *testing.T) {
+		r := doRPC(t, proc, apexRPCAddr, "header.LocalHead")
+		if r.Error != nil {
+			t.Fatalf("header.LocalHead error: %s", r.Error.Message)
+		}
+		if len(r.Result) == 0 || string(r.Result) == "null" {
+			t.Fatal("expected non-null local head")
+		}
+	})
+}

justfile

@@ -43,6 +43,9 @@ proto:
 # Run all checks (CI equivalent)
 check: tidy-check lint test build
 
-# Run the Docker-backed submission e2e test in the isolated e2e module.
-e2e-submission:
+# Run all Docker-backed e2e tests in the isolated e2e module (requires Docker).
+e2e:
     cd e2e && go test -race -count=1 -timeout 20m ./...
+
+# Run the full CI pipeline locally: lint + unit tests + build + e2e (requires Docker).
+ci: check e2e