dberkov
diff --git a/‎cmd/ateapi/internal/controlapi/functional_test.go‎
Lines changed: 88 additions & 10 deletions b/‎cmd/ateapi/internal/controlapi/functional_test.go‎
Lines changed: 88 additions & 10 deletions
diff --git a/‎cmd/ateapi/internal/controlapi/list_actors.go‎
Lines changed: 21 additions & 4 deletions b/‎cmd/ateapi/internal/controlapi/list_actors.go‎
Lines changed: 21 additions & 4 deletions
diff --git a/‎cmd/ateapi/internal/store/ateredis/ateredis.go‎
Lines changed: 159 additions & 22 deletions b/‎cmd/ateapi/internal/store/ateredis/ateredis.go‎
Lines changed: 159 additions & 22 deletions
@@ -20,7 +20,6 @@ import (
 	"net"
 	"os"
 	"os/exec"
-	"sort"
 	"strings"
 	"sync"
 	"testing"
@@ -36,6 +35,7 @@ import (
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"github.com/alicebob/miniredis/v2"
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/redis/go-redis/v9"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -619,23 +619,101 @@ func TestListActors(t *testing.T) {
 		t.Fatalf("expected 2 actors, got %d", len(listResp.Actors))
 	}
 
-	sort.Slice(listResp.Actors, func(i, j int) bool {
-		return listResp.Actors[i].ActorId < listResp.Actors[j].ActorId
-	})
-
 	want := []*ateapipb.Actor{
 		resp1.GetActor(),
 		resp2.GetActor(),
 	}
-	sort.Slice(want, func(i, j int) bool {
-		return want[i].ActorId < want[j].ActorId
-	})
 
-	if diff := cmp.Diff(want, listResp.Actors, protocmp.Transform()); diff != "" {
+	opts := []cmp.Option{
+		protocmp.Transform(),
+		cmpopts.SortSlices(func(a, b *ateapipb.Actor) bool {
+			return a.ActorId < b.ActorId
+		}),
+	}
+
+	if diff := cmp.Diff(want, listResp.Actors, opts...); diff != "" {
 		t.Errorf("ListActors response mismatch (-want +got):\n%s", diff)
 	}
 }
 
+// TestListActors_Pagination tests that ListActors correctly paginates results.
+func TestListActors_Pagination(t *testing.T) {
+	ns := namespaceForTest("ns-list-actors-pagination")
+	tc := setupTest(t, ns)
+	defer tc.cleanup()
+
+	createTemplate(t, tc, ns)
+
+	var want []*ateapipb.Actor
+	for i := 0; i < 5; i++ {
+		resp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
+			ActorTemplateNamespace: ns,
+			ActorTemplateName:      "tmpl1",
+			ActorId:                fmt.Sprintf("id%d", i),
+		})
+		if err != nil {
+			t.Fatalf("CreateActor %d failed: %v", i, err)
+		}
+		want = append(want, resp.GetActor())
+	}
+
+	var allActors []*ateapipb.Actor
+	pageToken := ""
+
+	for {
+		listResp, err := tc.client.ListActors(context.Background(), &ateapipb.ListActorsRequest{
+			PageSize:  2,
+			PageToken: pageToken,
+		})
+		if err != nil {
+			t.Fatalf("ListActors failed: %v", err)
+		}
+
+		allActors = append(allActors, listResp.Actors...)
+		pageToken = listResp.GetNextPageToken()
+		if pageToken == "" {
+			break
+		}
+	}
+
+	if len(allActors) != 5 {
+		t.Fatalf("expected 5 actors total, got %d", len(allActors))
+	}
+
+	opts := []cmp.Option{
+		protocmp.Transform(),
+		cmpopts.SortSlices(func(a, b *ateapipb.Actor) bool {
+			return a.ActorId < b.ActorId
+		}),
+	}
+
+	if diff := cmp.Diff(want, allActors, opts...); diff != "" {
+		t.Errorf("ListActors pagination response mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func TestListActors_PageSizeValidation(t *testing.T) {
+	ns := namespaceForTest("ns-list-actors-validation")
+	tc := setupTest(t, ns)
+	defer tc.cleanup()
+
+	// 1. Negative page size
+	_, err := tc.client.ListActors(context.Background(), &ateapipb.ListActorsRequest{
+		PageSize: -1,
+	})
+	if status.Code(err) != codes.InvalidArgument {
+		t.Errorf("expected InvalidArgument error for negative page_size, got: %v", err)
+	}
+
+	// 2. Page size exceeding maxPageSize (1000)
+	_, err = tc.client.ListActors(context.Background(), &ateapipb.ListActorsRequest{
+		PageSize: 1001,
+	})
+	if status.Code(err) != codes.InvalidArgument {
+		t.Errorf("expected InvalidArgument error for page_size > 1000, got: %v", err)
+	}
+}
+
 // TestListWorkers tests that workers mirrored to Redis are listed.
 // Workflow:
 // 1. Creates a mock worker Pod in Kubernetes.
@@ -1203,7 +1281,7 @@ func TestSuspendActor_DanglingWorker(t *testing.T) {
 	deleteWorkerPod(t, tc, ns, "worker-1")
 
 	// 3. Call SuspendActor -> Should succeed (our fix skips missing pod execution)
-	actors, _ := tc.persistence.ListActors(context.Background())
+	actors, _, _ := tc.persistence.ListActors(context.Background(), maxPageSize, "")
 	t.Logf("Actors in Redis before Suspend: %d", len(actors))
 	for _, a := range actors {
 		t.Logf("  Actor: %s/%s/%s", a.GetActorTemplateNamespace(), a.GetActorTemplateName(), a.GetActorId())
 
@@ -19,21 +19,38 @@ import (
 	"fmt"
 
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
+const maxPageSize = 1000
+
 func (s *Service) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest) (*ateapipb.ListActorsResponse, error) {
 	if err := validateListActorsRequest(req); err != nil {
-		return nil, err
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+	pageSize := req.GetPageSize()
+	if pageSize == 0 {
+		pageSize = maxPageSize
 	}
-	actors, err := s.persistence.ListActors(ctx)
+
+	actors, nextToken, err := s.persistence.ListActors(ctx, pageSize, req.GetPageToken())
 	if err != nil {
 		return nil, fmt.Errorf("while listing actors in db: %w", err)
 	}
 	return &ateapipb.ListActorsResponse{
-		Actors: actors,
+		Actors:        actors,
+		NextPageToken: nextToken,
 	}, nil
 }
 
-func validateListActorsRequest(_ *ateapipb.ListActorsRequest) error {
+func validateListActorsRequest(req *ateapipb.ListActorsRequest) error {
+	pageSize := req.GetPageSize()
+	if pageSize < 0 {
+		return fmt.Errorf("page_size cannot be negative")
+	}
+	if pageSize > maxPageSize {
+		return fmt.Errorf("page_size %d exceeds maximum page size %d", pageSize, maxPageSize)
+	}
 	return nil
 }
@@ -41,8 +41,13 @@ package ateredis
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -393,40 +398,172 @@ func (s *Persistence) ListWorkers(ctx context.Context) ([]*ateapipb.Worker, erro
 	return result, nil
 }
 
-func (s *Persistence) ListActors(ctx context.Context) ([]*ateapipb.Actor, error) {
+type listActorsPageToken struct {
+	ShardHash string `json:"shard_hash"`
+	Cursor    uint64 `json:"cursor"`
+}
+
+func encodePageToken(token listActorsPageToken) string {
+	b, _ := json.Marshal(token)
+	return base64.StdEncoding.EncodeToString(b)
+}
+
+func decodePageToken(tokenStr string) (listActorsPageToken, error) {
+	var token listActorsPageToken
+	if tokenStr == "" {
+		return token, nil
+	}
+	b, err := base64.StdEncoding.DecodeString(tokenStr)
+	if err != nil {
+		return token, err
+	}
+	err = json.Unmarshal(b, &token)
+	return token, err
+}
+
+func hashShardAddr(addr string) string {
+	h := sha256.Sum256([]byte(addr))
+	return hex.EncodeToString(h[:])
+}
+
+func (s *Persistence) ListActors(ctx context.Context, pageSize int32, pageTokenStr string) ([]*ateapipb.Actor, string, error) {
+	token, err := decodePageToken(pageTokenStr)
+	if err != nil {
+		return nil, "", fmt.Errorf("invalid page token: %w", err)
+	}
+
+	masters, err := s.getSortedMasters(ctx)
+	if err != nil {
+		return nil, "", err
+	}
+
+	startIndex, err := findStartingShard(masters, token.ShardHash)
+	if err != nil {
+		return nil, "", err
+	}
+
 	var result []*ateapipb.Actor
-	var mu sync.Mutex
+	var nextToken string
+	stop := false
 
-	err := s.rdb.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
-		iter := master.Scan(ctx, 0, "actor:*", 0).Iterator()
-		for iter.Next(ctx) {
-			actorKey := iter.Val()
-			parts := strings.Split(actorKey, ":")
-			if len(parts) != 2 {
-				return fmt.Errorf("bad key format %q", actorKey)
+	for i := startIndex; i < len(masters) && !stop; i++ {
+		master := masters[i]
+		shardAddr := master.Options().Addr
+		cursor := uint64(0)
+		if i == startIndex && token.ShardHash != "" {
+			cursor = token.Cursor
+		}
+
+		for {
+			remaining := int(pageSize) - len(result)
+			if remaining <= 0 {
+				if cursor != 0 {
+					nextToken = encodePageToken(listActorsPageToken{
+						ShardHash: hashShardAddr(shardAddr),
+						Cursor:    cursor,
+					})
+				} else if i+1 < len(masters) {
+					nextToken = encodePageToken(listActorsPageToken{
+						ShardHash: hashShardAddr(masters[i+1].Options().Addr),
+						Cursor:    0,
+					})
+				} else {
+					nextToken = ""
+				}
+				stop = true
+				break
 			}
 
-			getCmd := master.Get(ctx, actorKey)
-			if getCmd.Err() != nil {
-				return fmt.Errorf("while getting actor %q: %w", actorKey, getCmd.Err())
+			var keys []string
+			keys, cursor, err = master.Scan(ctx, cursor, "actor:*", int64(remaining)).Result()
+			if err != nil {
+				return nil, "", fmt.Errorf("while scanning shard %s: %w", shardAddr, err)
 			}
 
-			actor := &ateapipb.Actor{}
-			if err := protojson.Unmarshal([]byte(getCmd.Val()), actor); err != nil {
-				return fmt.Errorf("in protojson.Unmarshal: %w", err)
+			if len(keys) > 0 {
+				actors, err := s.fetchActors(ctx, master, keys)
+				if err != nil {
+					return nil, "", err
+				}
+				result = append(result, actors...)
 			}
 
-			mu.Lock()
-			result = append(result, actor)
-			mu.Unlock()
+			if cursor == 0 {
+				if i+1 < len(masters) {
+					nextToken = encodePageToken(listActorsPageToken{
+						ShardHash: hashShardAddr(masters[i+1].Options().Addr),
+						Cursor:    0,
+					})
+				} else {
+					nextToken = ""
+				}
+				break
+			}
 		}
-		return iter.Err()
-	})
+	}
+
+	return result, nextToken, nil
+}
 
+func (s *Persistence) getSortedMasters(ctx context.Context) ([]*redis.Client, error) {
+	var masters []*redis.Client
+	err := s.rdb.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
+		masters = append(masters, master)
+		return nil
+	})
 	if err != nil {
-		return nil, fmt.Errorf("while iterating all redis master: %w", err)
+		return nil, fmt.Errorf("while listing redis masters: %w", err)
 	}
-	return result, nil
+
+	sort.Slice(masters, func(i, j int) bool {
+		return masters[i].Options().Addr < masters[j].Options().Addr
+	})
+	return masters, nil
+}
+
+func findStartingShard(masters []*redis.Client, shardHash string) (int, error) {
+	if shardHash == "" {
+		return 0, nil
+	}
+	for i, m := range masters {
+		if hashShardAddr(m.Options().Addr) == shardHash {
+			return i, nil
+		}
+	}
+	return 0, fmt.Errorf("topology changed: shard with hash %s not found (aborted)", shardHash)
+}
+
+func (s *Persistence) fetchActors(ctx context.Context, master *redis.Client, keys []string) ([]*ateapipb.Actor, error) {
+	cmds, err := master.Pipelined(ctx, func(pipe redis.Pipeliner) error {
+		for _, key := range keys {
+			pipe.Get(ctx, key)
+		}
+		return nil
+	})
+	if err != nil && !errors.Is(err, redis.Nil) {
+		return nil, fmt.Errorf("while fetching keys in shard %s: %w", master.Options().Addr, err)
+	}
+
+	var actors []*ateapipb.Actor
+	for _, cmd := range cmds {
+		getCmd, ok := cmd.(*redis.StringCmd)
+		if !ok {
+			continue
+		}
+		if getCmd.Err() != nil {
+			if errors.Is(getCmd.Err(), redis.Nil) {
+				continue
+			}
+			return nil, fmt.Errorf("while getting actor: %w", getCmd.Err())
+		}
+
+		actor := &ateapipb.Actor{}
+		if err := protojson.Unmarshal([]byte(getCmd.Val()), actor); err != nil {
+			return nil, fmt.Errorf("in protojson.Unmarshal: %w", err)
+		}
+		actors = append(actors, actor)
+	}
+	return actors, nil
 }
 
 func (s *Persistence) AcquireLock(ctx context.Context, key string, value string, ttl time.Duration) (bool, error) {