fix: fetch schema over network to ensure @unique check across groups (#9596)

**Description**

This PR fixes an issue where the `@unique` constraint is bypassed in
multi-group Dgraph environments. This occurs when a mutation is
processed by an Alpha node that does not own the predicate, as the local
schema state lacks the necessary metadata for "remote" predicates.(see
#9565) Closes #9565

To address this, I've implemented an on-demand fetching mechanism:
- **On-Demand Fetching**: If a predicate is missing from the local
`schema.State`, the Alpha node now fetches the schema over the network
from the owner group using `worker.GetSchemaOverNetwork`.
- **Per-Request Cache**: Introduced a `repaired` map within
`addQueryIfUnique` to cache these network lookups for the duration of
the request. This prevents RPC amplification and ensures performance
remains stable during large mutations.
- **Scope Isolation**: The fetched metadata is only used for the current
request context, preventing any potential stale data issues in the
global schema state.

This ensures that the `@unique` constraint is correctly validated
cluster-wide, regardless of which group handles the mutation.

**Checklist**

- [x] The PR title follows the
[Conventional
Commits](https://www.conventionalcommits.org/en/v1.0.0/#summary) syntax,
leading
      with `fix:`, `feat:`, `chore:`, `ci:`, etc.
- [x] Code compiles correctly and linting (via trunk) passes locally
- [x] Tests added for new functionality, or regression tests for bug
fixes added as applicable

---------

Co-authored-by: Shiva <shiva@Shivajis-MacBook-Pro.local>

Author: eileenaaa

dgraphapi/cluster.go

@@ -63,6 +63,7 @@ type HTTPClient struct {
 	dqlURL        string
 	dqlMutateUrl  string
 	alphaStateUrl string
+	moveTabletURL string
 }
 
 // GraphQLParams are used for making graphql requests to dgraph
@@ -696,6 +697,30 @@ func (hc *HTTPClient) Mutate(mutation string, commitNow bool) ([]byte, error) {
 	return DoReq(req)
 }
 
+func (hc *HTTPClient) MoveTablet(predicate string, group uint32) error {
+	url := fmt.Sprintf("%s?tablet=%s&group=%d", hc.moveTabletURL, predicate, group)
+	response, err := http.Get(url)
+	if err != nil {
+		return errors.Wrapf(err, "error moving tablet via HTTP: predicate=%s, group=%d", predicate, group)
+	}
+	defer func() {
+		if err := response.Body.Close(); err != nil {
+			log.Printf("[WARNING] error closing body: %v", err)
+		}
+	}()
+
+	body, err := io.ReadAll(response.Body)
+	if err != nil {
+		return errors.Wrapf(err, "error reading move tablet response body")
+	}
+
+	if response.StatusCode != http.StatusOK {
+		return errors.Errorf("move tablet failed with status %d: %s", response.StatusCode, string(body))
+	}
+
+	return nil
+}
+
 // SetupSchema sets up DQL schema
 func (gc *GrpcClient) SetupSchema(dbSchema string) error {
 	ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
@@ -784,9 +809,11 @@ func GetHttpClient(alphaUrl, zeroUrl string) (*HTTPClient, error) {
 
 	dqlUrl := "http://" + alphaUrl + "/query"
 	dqlMutateUrl := "http://" + alphaUrl + "/mutate"
+	moveTabletUrl := "http://" + zeroUrl + "/moveTablet"
 	return &HTTPClient{
 		adminURL:      adminUrl,
 		graphqlURL:    graphQLUrl,
+		moveTabletURL: moveTabletUrl,
 		stateURL:      stateUrl,
 		dqlURL:        dqlUrl,
 		dqlMutateUrl:  dqlMutateUrl,

edgraph/server.go

@@ -1693,6 +1693,41 @@ func addQueryIfUnique(qctx context.Context, qc *queryContext) error {
 	}
 	isGalaxyQuery := x.IsRootNsOperation(ctx)
 
+	missingPreds := make(map[string]struct{})
+	for _, gmu := range qc.gmuList {
+		for _, pred := range gmu.Set {
+			currNs := namespace
+			if isGalaxyQuery {
+				currNs = pred.Namespace
+			}
+			if pred.Predicate == "dgraph.xid" {
+				continue
+			}
+			fullPred := x.NamespaceAttr(currNs, pred.Predicate)
+			if _, ok := schema.State().Get(ctx, fullPred); !ok {
+				missingPreds[fullPred] = struct{}{}
+			}
+		}
+	}
+
+	repaired := make(map[string]bool)
+	if len(missingPreds) > 0 {
+		predList := make([]string, 0, len(missingPreds))
+		for p := range missingPreds {
+			predList = append(predList, p)
+		}
+
+		schReq := &pb.SchemaRequest{Predicates: predList}
+		remoteNodes, err := worker.GetSchemaOverNetwork(ctx, schReq)
+		if err != nil {
+			return errors.Wrapf(err, "unique validation failed to fetch schema for predicates %v", predList)
+		}
+
+		for _, node := range remoteNodes {
+			repaired[node.Predicate] = node.Unique
+		}
+	}
+
 	qc.uniqueVars = map[uint64]uniquePredMeta{}
 	for gmuIndex, gmu := range qc.gmuList {
 		var buildQuery strings.Builder
@@ -1706,7 +1741,16 @@ func addQueryIfUnique(qctx context.Context, qc *queryContext) error {
 				// [TODO] Don't check if it's dgraph.xid. It's a bug as this node might not be aware
 				// of the schema for the given predicate. This is a bug issue for dgraph.xid hence
 				// we are bypassing it manually until the bug is fixed.
-				predSchema, ok := schema.State().Get(ctx, x.NamespaceAttr(namespace, pred.Predicate))
+				fullPred := x.NamespaceAttr(namespace, pred.Predicate)
+				predSchema, ok := schema.State().Get(ctx, fullPred)
+				if !ok {
+					u, found := repaired[fullPred]
+					if found {
+						predSchema.Unique = u
+						ok = true
+					}
+				}
+
 				if !ok || !predSchema.Unique {
 					continue
 				}

systest/integration2/directive_test.go

@@ -0,0 +1,158 @@
+//go:build integration2
+
+/*
+ * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package main
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/dgraph-io/dgo/v250/protos/api"
+	"github.com/dgraph-io/dgraph/v25/dgraphtest"
+	"github.com/dgraph-io/dgraph/v25/protos/pb"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+func TestUniqueMultipleGroups(t *testing.T) {
+	conf := dgraphtest.NewClusterConfig().WithNumAlphas(2).WithNumZeros(1).WithReplicas(1)
+	c, err := dgraphtest.NewLocalCluster(conf)
+	require.NoError(t, err)
+	defer func() { c.Cleanup(t.Failed()) }()
+	require.NoError(t, c.Start())
+
+	httpClient, err := c.HTTPClient()
+	require.NoError(t, err)
+
+	dg, cleanup, err := c.Client()
+	require.NoError(t, err)
+	defer cleanup()
+
+	require.NoError(t, dg.DropAll())
+
+	// Setup schema
+	schema := `
+        name: string @index(exact) .
+        email_group_1: string @unique @index(exact) .
+        email_group_2: string @unique @index(exact) .
+    `
+	require.NoError(t, dg.SetupSchema(schema))
+	time.Sleep(2 * time.Second)
+
+	// Get the cluster state to find which alphas belong to which groups
+	var state pb.MembershipState
+
+	getStateAndFindGroups := func() (uint32, uint32) {
+		healthResp, err := httpClient.GetAlphaState()
+		require.NoError(t, err)
+		require.NoError(t, protojson.Unmarshal(healthResp, &state))
+		t.Logf("Retrieved cluster state with %d groups", len(state.Groups))
+		var e1, e2 uint32
+		for groupID, group := range state.Groups {
+			for predName := range group.Tablets {
+
+				if predName == "email_group_1" || predName == "0-email_group_1" {
+					t.Logf("predName: %s", predName)
+					t.Logf("groupID: %d", groupID)
+					e1 = groupID
+				}
+				if predName == "email_group_2" || predName == "0-email_group_2" {
+					t.Logf("predName: %s", predName)
+					t.Logf("groupID: %d", groupID)
+					e2 = groupID
+				}
+			}
+		}
+		return e1, e2
+	}
+
+	email1GroupID, email2GroupID := getStateAndFindGroups()
+	require.NotZero(t, email1GroupID, "email_group_1 predicate should be assigned to a group")
+	require.NotZero(t, email2GroupID, "email_group_2 predicate should be assigned to a group")
+
+	// If both predicates are in the same group, move one to a different group
+	if email1GroupID == email2GroupID {
+		var targetGroupID uint32
+		for gid := range state.Groups {
+			if gid != email1GroupID && gid > 0 {
+				targetGroupID = gid
+				break
+			}
+		}
+		require.NotZero(t, targetGroupID, "need at least two groups to move predicate")
+		err = httpClient.MoveTablet("email_group_1", targetGroupID)
+		if err != nil {
+			require.Contains(t, err.Error(), "already being served by group")
+		}
+		t.Logf("Moved email_group_1 predicate to group %d (was in same group as email_group_2)", targetGroupID)
+		time.Sleep(2 * time.Second)
+		email1GroupID, email2GroupID = getStateAndFindGroups()
+	}
+
+	require.NotEqual(t, email1GroupID, email2GroupID, "email_group_1 and email_group_2 should be in different groups")
+
+	t.Logf("email_group_1 is in group %d, email_group_2 is in group %d", email1GroupID, email2GroupID)
+
+	// Helper to find an alpha index in a given group
+	findAlphaInGroup := func(gid uint32) int {
+		if group, ok := state.Groups[gid]; ok {
+			for _, member := range group.Members {
+				addr := member.Addr
+				if strings.HasPrefix(addr, "alpha") {
+					alphaNumStr := strings.TrimPrefix(addr, "alpha")
+					alphaNumStr = strings.Split(alphaNumStr, ":")[0]
+					alphaNum, err := strconv.Atoi(alphaNumStr)
+					if err == nil && alphaNum >= 0 {
+						return alphaNum
+					}
+				}
+			}
+		}
+		return -1
+	}
+	alphaForEmailGroup1 := findAlphaInGroup(email1GroupID)
+	alphaForEmailGroup2 := findAlphaInGroup(email2GroupID)
+	require.GreaterOrEqual(t, alphaForEmailGroup1, 0, "should find an alpha serving email_group_1")
+	require.GreaterOrEqual(t, alphaForEmailGroup2, 0, "should find an alpha serving email_group_2")
+
+	// Insert initial values for both predicates using default client
+	_, err = dg.Mutate(&api.Mutation{
+		SetNquads: []byte(`<0x100> <name> "User One" .
+             <0x100> <email_group_1> "dup@example.com" .
+             <0x100> <email_group_2> "other@example.com" .`),
+		CommitNow: true,
+	})
+	require.NoError(t, err)
+
+	// Try to insert duplicate into email1 from an alpha that does NOT serve email1 (i.e. from email2's group)
+	dgNotEmail1, cleanup1, err := c.AlphaClient(alphaForEmailGroup2)
+	require.NoError(t, err)
+	defer cleanup1()
+	rdfDupEmail1 := `<0x200> <name> "User Two" .
+             <0x200> <email_group_1> "dup@example.com" .`
+	_, err = dgNotEmail1.Mutate(&api.Mutation{
+		SetNquads: []byte(rdfDupEmail1),
+		CommitNow: true,
+	})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "could not insert duplicate value")
+
+	// Try to insert duplicate into email2 from an alpha that does NOT serve email2 (i.e. from email1's group)
+	dgNotEmail2, cleanup2, err := c.AlphaClient(alphaForEmailGroup1)
+	require.NoError(t, err)
+	defer cleanup2()
+	rdfDupEmail2 := `<0x300> <name> "User Three" .
+             <0x300> <email_group_2> "other@example.com" .`
+	_, err = dgNotEmail2.Mutate(&api.Mutation{
+		SetNquads: []byte(rdfDupEmail2),
+		CommitNow: true,
+	})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "could not insert duplicate value")
+}