From a958887ff62e0ffa61f09b6fab100a1ed75b0397 Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Fri, 10 Jul 2026 11:34:42 -0700 Subject: [PATCH] Rework List endpoints to align with API style guide. --- .../internal/controlapi/functional_test.go | 67 +++++- cmd/ateapi/internal/controlapi/list_actors.go | 41 ++-- .../internal/controlapi/list_atespaces.go | 28 ++- .../internal/controlapi/list_workers.go | 22 +- cmd/ateapi/internal/controlapi/syncer_test.go | 2 +- .../internal/store/ateredis/ateredis.go | 142 +++++------ .../internal/store/ateredis/ateredis_test.go | 220 +++++++++++++++++- cmd/ateapi/internal/store/store.go | 8 +- .../internal/workercache/workercache.go | 18 +- .../internal/workercache/workercache_test.go | 6 +- cmd/kubectl-ate/internal/cmd/get_atespaces.go | 21 +- cmd/kubectl-ate/internal/cmd/get_workers.go | 21 +- pkg/proto/ateapipb/ateapi.pb.go | 142 ++++++++--- pkg/proto/ateapipb/ateapi.proto | 63 +++-- pkg/proto/ateapipb/ateapi_grpc.pb.go | 12 +- 15 files changed, 620 insertions(+), 193 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 57c08a9c5..ac35b3897 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -1061,13 +1061,12 @@ func TestListActors_PageSizeValidation(t *testing.T) { 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{ + // 2. Page size exceeding maxPageSize (1000) is coerced silently. + if _, err := tc.client.ListActors(context.Background(), &ateapipb.ListActorsRequest{ Atespace: testAtespace, PageSize: 1001, - }) - if status.Code(err) != codes.InvalidArgument { - t.Errorf("expected InvalidArgument error for page_size > 1000, got: %v", err) + }); err != nil { + t.Errorf("expected page_size > 1000 to be coerced, got error: %v", err) } } @@ -2295,6 +2294,64 @@ func TestValidation(t *testing.T) { }) } }) + + t.Run("ListActors", func(t *testing.T) { + tests := []struct { + name string + req *ateapipb.ListActorsRequest + wantMsg string + }{{ + "invalid atespace", + &ateapipb.ListActorsRequest{Atespace: "NS1"}, + "atespace: Invalid value", + }, { + "negative page_size", + &ateapipb.ListActorsRequest{Atespace: "ns1", PageSize: -1}, + "page_size: Invalid value", + }} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := tc.client.ListActors(context.Background(), tt.req) + assertGrpcErrorRegex(t, err, codes.InvalidArgument, tt.wantMsg) + }) + } + }) + + t.Run("ListWorkers", func(t *testing.T) { + tests := []struct { + name string + req *ateapipb.ListWorkersRequest + wantMsg string + }{{ + "negative page_size", + &ateapipb.ListWorkersRequest{PageSize: -1}, + "page_size: Invalid value", + }} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := tc.client.ListWorkers(context.Background(), tt.req) + assertGrpcErrorRegex(t, err, codes.InvalidArgument, tt.wantMsg) + }) + } + }) + + t.Run("ListAtespaces", func(t *testing.T) { + tests := []struct { + name string + req *ateapipb.ListAtespacesRequest + wantMsg string + }{{ + "negative page_size", + &ateapipb.ListAtespacesRequest{PageSize: -1}, + "page_size: Invalid value", + }} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := tc.client.ListAtespaces(context.Background(), tt.req) + assertGrpcErrorRegex(t, err, codes.InvalidArgument, tt.wantMsg) + }) + } + }) } func TestResumeActor_LockConflict(t *testing.T) { diff --git a/cmd/ateapi/internal/controlapi/list_actors.go b/cmd/ateapi/internal/controlapi/list_actors.go index e22ba51c3..202033085 100644 --- a/cmd/ateapi/internal/controlapi/list_actors.go +++ b/cmd/ateapi/internal/controlapi/list_actors.go @@ -22,20 +22,26 @@ import ( "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/util/validation/field" ) const maxPageSize = 1000 +// effectivePageSize applies the server-chosen default for an unset page_size +// and silently coerces oversized values. +func effectivePageSize(requested int32) int32 { + if requested == 0 || requested > maxPageSize { + return maxPageSize + } + return requested +} + func (s *Service) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest) (*ateapipb.ListActorsResponse, error) { if err := validateListActorsRequest(req); err != nil { - return nil, status.Error(codes.InvalidArgument, err.Error()) - } - pageSize := req.GetPageSize() - if pageSize == 0 { - pageSize = maxPageSize + return nil, err } - actors, nextToken, err := s.persistence.ListActors(ctx, req.GetAtespace(), pageSize, req.GetPageToken()) + actors, nextToken, err := s.persistence.ListActors(ctx, req.GetAtespace(), effectivePageSize(req.GetPageSize()), req.GetPageToken()) if err != nil { return nil, fmt.Errorf("while listing actors in db: %w", err) } @@ -46,17 +52,22 @@ func (s *Service) ListActors(ctx context.Context, req *ateapipb.ListActorsReques } func validateListActorsRequest(req *ateapipb.ListActorsRequest) error { - // An empty atespace is allowed here and means "all atespaces"(used by `kubectl ate get actors -A`). - // A non-empty atespace is validated and scopes the listing to that atespace. - if req.GetAtespace() != "" && !resources.IsValidResourceName(req.GetAtespace()) { - return fmt.Errorf("invalid atespace %q: must be a valid resource name", req.GetAtespace()) + var fldPath *field.Path + var errs field.ErrorList + + // An empty atespace is allowed here and means "all atespaces" (used by + // `kubectl ate get actors -A`). A non-empty atespace is validated and + // scopes the listing to that atespace. + if val, fldPath := req.Atespace, fldPath.Child("atespace"); val != "" { + errs = append(errs, resources.ValidateResourceName(val, fldPath)...) } - pageSize := req.GetPageSize() - if pageSize < 0 { - return fmt.Errorf("page_size cannot be negative") + + if val, fldPath := req.PageSize, fldPath.Child("page_size"); val < 0 { + errs = append(errs, field.Invalid(fldPath, val, "must be greater than or equal to 0")) } - if pageSize > maxPageSize { - return fmt.Errorf("page_size %d exceeds maximum page size %d", pageSize, maxPageSize) + + if len(errs) > 0 { + return status.Error(codes.InvalidArgument, errs.ToAggregate().Error()) } return nil } diff --git a/cmd/ateapi/internal/controlapi/list_atespaces.go b/cmd/ateapi/internal/controlapi/list_atespaces.go index 27b15e31e..68620e281 100644 --- a/cmd/ateapi/internal/controlapi/list_atespaces.go +++ b/cmd/ateapi/internal/controlapi/list_atespaces.go @@ -19,12 +19,36 @@ import ( "fmt" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/util/validation/field" ) func (s *Service) ListAtespaces(ctx context.Context, req *ateapipb.ListAtespacesRequest) (*ateapipb.ListAtespacesResponse, error) { - atespaces, err := s.persistence.ListAtespaces(ctx) + if err := validateListAtespacesRequest(req); err != nil { + return nil, err + } + + atespaces, nextToken, err := s.persistence.ListAtespaces(ctx, effectivePageSize(req.GetPageSize()), req.GetPageToken()) if err != nil { return nil, fmt.Errorf("while listing atespaces in db: %w", err) } - return &ateapipb.ListAtespacesResponse{Atespaces: atespaces}, nil + return &ateapipb.ListAtespacesResponse{ + Atespaces: atespaces, + NextPageToken: nextToken, + }, nil +} + +func validateListAtespacesRequest(req *ateapipb.ListAtespacesRequest) error { + var fldPath *field.Path + var errs field.ErrorList + + if val, fldPath := req.PageSize, fldPath.Child("page_size"); val < 0 { + errs = append(errs, field.Invalid(fldPath, val, "must be greater than or equal to 0")) + } + + if len(errs) > 0 { + return status.Error(codes.InvalidArgument, errs.ToAggregate().Error()) + } + return nil } diff --git a/cmd/ateapi/internal/controlapi/list_workers.go b/cmd/ateapi/internal/controlapi/list_workers.go index 1e7aa7650..6a437cd7e 100644 --- a/cmd/ateapi/internal/controlapi/list_workers.go +++ b/cmd/ateapi/internal/controlapi/list_workers.go @@ -19,22 +19,36 @@ import ( "fmt" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/util/validation/field" ) func (s *Service) ListWorkers(ctx context.Context, req *ateapipb.ListWorkersRequest) (*ateapipb.ListWorkersResponse, error) { if err := validateListWorkersRequest(req); err != nil { return nil, err } - workers, err := s.persistence.ListWorkers(ctx) + + workers, nextToken, err := s.persistence.ListWorkers(ctx, effectivePageSize(req.GetPageSize()), req.GetPageToken()) if err != nil { return nil, fmt.Errorf("while listing workers in db: %w", err) } return &ateapipb.ListWorkersResponse{ - Workers: workers, + Workers: workers, + NextPageToken: nextToken, }, nil } -func validateListWorkersRequest(_ *ateapipb.ListWorkersRequest) error { - // No fields to validate for now. +func validateListWorkersRequest(req *ateapipb.ListWorkersRequest) error { + var fldPath *field.Path + var errs field.ErrorList + + if val, fldPath := req.PageSize, fldPath.Child("page_size"); val < 0 { + errs = append(errs, field.Invalid(fldPath, val, "must be greater than or equal to 0")) + } + + if len(errs) > 0 { + return status.Error(codes.InvalidArgument, errs.ToAggregate().Error()) + } return nil } diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 8e1985801..6452bfaf3 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -88,7 +88,7 @@ func TestSyncer_Lifecycle(t *testing.T) { defer cleanup() // 1. Verify no workers in Redis initially - workers, err := persistence.ListWorkers(context.Background()) + workers, _, err := persistence.ListWorkers(context.Background(), 1000, "") if err != nil { t.Fatalf("failed to list workers: %v", err) } diff --git a/cmd/ateapi/internal/store/ateredis/ateredis.go b/cmd/ateapi/internal/store/ateredis/ateredis.go index 139617187..03140c6a6 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis.go @@ -49,7 +49,6 @@ import ( "fmt" "log/slog" "sort" - "strings" "sync" "time" @@ -156,35 +155,20 @@ func (s *Persistence) AtespaceExists(ctx context.Context, name string) (bool, er return n > 0, nil } -func (s *Persistence) ListAtespaces(ctx context.Context) ([]*ateapipb.Atespace, error) { +func (s *Persistence) ListAtespaces(ctx context.Context, pageSize int32, pageTokenStr string) ([]*ateapipb.Atespace, string, error) { var result []*ateapipb.Atespace - var mu sync.Mutex - - err := s.rdb.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error { - iter := master.Scan(ctx, 0, "atespace:*", 0).Iterator() - for iter.Next(ctx) { - key := iter.Val() - getCmd := master.Get(ctx, key) - if getCmd.Err() != nil { - return fmt.Errorf("while getting atespace %q: %w", key, getCmd.Err()) - } - atespace := &ateapipb.Atespace{} - if err := protojson.Unmarshal([]byte(getCmd.Val()), atespace); err != nil { - return fmt.Errorf("in protojson.Unmarshal: %w", err) - } - mu.Lock() - result = append(result, atespace) - mu.Unlock() - } - if err := iter.Err(); err != nil { - return fmt.Errorf("error from iterator: %w", err) + nextToken, err := s.listPage(ctx, "atespace:*", pageSize, pageTokenStr, func(ctx context.Context, master *redis.Client, keys []string) (int, error) { + atespaces, err := fetchProtos(ctx, master, keys, func() *ateapipb.Atespace { return &ateapipb.Atespace{} }) + if err != nil { + return 0, err } - return nil + result = append(result, atespaces...) + return len(atespaces), nil }) if err != nil { - return nil, fmt.Errorf("while iterating all redis master: %w", err) + return nil, "", err } - return result, nil + return result, nextToken, nil } // DeleteAtespace deletes an empty atespace. Returns store.ErrNotFound if the @@ -599,58 +583,34 @@ func (s *Persistence) UpdateActor(ctx context.Context, actor *ateapipb.Actor, ex return dbActor, nil } -func (s *Persistence) ListWorkers(ctx context.Context) ([]*ateapipb.Worker, error) { +func (s *Persistence) ListWorkers(ctx context.Context, pageSize int32, pageTokenStr string) ([]*ateapipb.Worker, string, error) { var result []*ateapipb.Worker - var mu sync.Mutex - - // Iterate through every Primary (Master) node in the cluster - err := s.rdb.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error { - iter := master.Scan(ctx, 0, "worker:*", 0).Iterator() - for iter.Next(ctx) { - workerKey := iter.Val() - parts := strings.Split(workerKey, ":") - if len(parts) != 4 { - return fmt.Errorf("bad key format %q", workerKey) - } - - getCmd := master.Get(ctx, workerKey) - if getCmd.Err() != nil { - return fmt.Errorf("while getting worker %q: %w", workerKey, getCmd.Err()) - } - - worker := &ateapipb.Worker{} - if err := protojson.Unmarshal([]byte(getCmd.Val()), worker); err != nil { - return fmt.Errorf("in protojson.Unmarshal: %w", err) - } - - mu.Lock() - result = append(result, worker) - mu.Unlock() - } - if err := iter.Err(); err != nil { - return fmt.Errorf("error from iterator: %w", err) + nextToken, err := s.listPage(ctx, "worker:*", pageSize, pageTokenStr, func(ctx context.Context, master *redis.Client, keys []string) (int, error) { + workers, err := fetchProtos(ctx, master, keys, func() *ateapipb.Worker { return &ateapipb.Worker{} }) + if err != nil { + return 0, err } - - return nil + result = append(result, workers...) + return len(workers), nil }) if err != nil { - return nil, fmt.Errorf("while iterating all redis master: %w", err) + return nil, "", err } - return result, nil + return result, nextToken, nil } -type listActorsPageToken struct { +type pageToken struct { ShardHash string `json:"shard_hash"` Cursor uint64 `json:"cursor"` } -func encodePageToken(token listActorsPageToken) string { +func encodePageToken(token pageToken) string { b, _ := json.Marshal(token) return base64.StdEncoding.EncodeToString(b) } -func decodePageToken(tokenStr string) (listActorsPageToken, error) { - var token listActorsPageToken +func decodePageToken(tokenStr string) (pageToken, error) { + var token pageToken if tokenStr == "" { return token, nil } @@ -671,41 +631,58 @@ func hashShardAddr(addr string) string { // across all atespaces (SCAN actor:*); a non-empty atespace restricts the scan to // that atespace (SCAN actor::*). func (s *Persistence) ListActors(ctx context.Context, atespace string, pageSize int32, pageTokenStr string) ([]*ateapipb.Actor, string, error) { + var result []*ateapipb.Actor + nextToken, err := s.listPage(ctx, actorScanPattern(atespace), pageSize, pageTokenStr, func(ctx context.Context, master *redis.Client, keys []string) (int, error) { + actors, err := fetchProtos(ctx, master, keys, func() *ateapipb.Actor { return &ateapipb.Actor{} }) + if err != nil { + return 0, err + } + result = append(result, actors...) + return len(actors), nil + }) + if err != nil { + return nil, "", err + } + return result, nextToken, nil +} + +// listPage SCANs pattern across the redis masters from the page token, feeding key batches to collect and returns the next-page token. +func (s *Persistence) listPage(ctx context.Context, pattern string, pageSize int32, pageTokenStr string, collect func(ctx context.Context, master *redis.Client, keys []string) (int, error)) (string, error) { token, err := decodePageToken(pageTokenStr) if err != nil { - return nil, "", fmt.Errorf("invalid page token: %w", err) + return "", fmt.Errorf("invalid page token: %w", err) } masters, err := s.getSortedMasters(ctx) if err != nil { - return nil, "", err + return "", err } startIndex, err := findStartingShard(masters, token.ShardHash) if err != nil { - return nil, "", err + return "", err } - var result []*ateapipb.Actor i := startIndex cursor := token.Cursor + collected := 0 - for i < len(masters) && len(result) < int(pageSize) { + for i < len(masters) && collected < int(pageSize) { master := masters[i] - remaining := int(pageSize) - len(result) + remaining := int(pageSize) - collected var keys []string - keys, cursor, err = master.Scan(ctx, cursor, actorScanPattern(atespace), int64(remaining)).Result() + keys, cursor, err = master.Scan(ctx, cursor, pattern, int64(remaining)).Result() if err != nil { - return nil, "", fmt.Errorf("while scanning shard %s: %w", master.Options().Addr, err) + return "", fmt.Errorf("while scanning shard %s: %w", master.Options().Addr, err) } if len(keys) > 0 { - actors, err := s.fetchActors(ctx, master, keys) + n, err := collect(ctx, master, keys) if err != nil { - return nil, "", err + return "", err } - result = append(result, actors...) + collected += n } if cursor == 0 { @@ -715,13 +692,13 @@ func (s *Persistence) ListActors(ctx context.Context, atespace string, pageSize var nextToken string if i < len(masters) { - nextToken = encodePageToken(listActorsPageToken{ + nextToken = encodePageToken(pageToken{ ShardHash: hashShardAddr(masters[i].Options().Addr), Cursor: cursor, }) } - return result, nextToken, nil + return nextToken, nil } func (s *Persistence) getSortedMasters(ctx context.Context) ([]*redis.Client, error) { @@ -756,7 +733,8 @@ func findStartingShard(masters []*redis.Client, shardHash string) (int, error) { 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) { +// fetchProtos fetches keys into newMsg-created messages. +func fetchProtos[M proto.Message](ctx context.Context, master *redis.Client, keys []string, newMsg func() M) ([]M, error) { cmds, err := master.Pipelined(ctx, func(pipe redis.Pipeliner) error { for _, key := range keys { pipe.Get(ctx, key) @@ -767,7 +745,7 @@ func (s *Persistence) fetchActors(ctx context.Context, master *redis.Client, key return nil, fmt.Errorf("while fetching keys in shard %s: %w", master.Options().Addr, err) } - var actors []*ateapipb.Actor + var out []M for _, cmd := range cmds { getCmd, ok := cmd.(*redis.StringCmd) if !ok { @@ -777,16 +755,16 @@ func (s *Persistence) fetchActors(ctx context.Context, master *redis.Client, key if errors.Is(getCmd.Err(), redis.Nil) { continue } - return nil, fmt.Errorf("while getting actor: %w", getCmd.Err()) + return nil, fmt.Errorf("while getting key: %w", getCmd.Err()) } - actor := &ateapipb.Actor{} - if err := protojson.Unmarshal([]byte(getCmd.Val()), actor); err != nil { + msg := newMsg() + if err := protojson.Unmarshal([]byte(getCmd.Val()), msg); err != nil { return nil, fmt.Errorf("in protojson.Unmarshal: %w", err) } - actors = append(actors, actor) + out = append(out, msg) } - return actors, nil + return out, nil } func (s *Persistence) AcquireLock(ctx context.Context, key string, value string, ttl time.Duration) (bool, error) { diff --git a/cmd/ateapi/internal/store/ateredis/ateredis_test.go b/cmd/ateapi/internal/store/ateredis/ateredis_test.go index 7e428d853..57135f3dc 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis_test.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis_test.go @@ -467,7 +467,7 @@ func TestListWorkers(t *testing.T) { t.Fatalf("failed to create worker2: %v", err) } - workers, err := s.ListWorkers(ctx) + workers, _, err := s.ListWorkers(ctx, 1000, "") if err != nil { t.Fatalf("ListWorkers failed: %v", err) } @@ -621,7 +621,7 @@ func TestListWorkers_Empty(t *testing.T) { mr, s, ctx := setupTest(t) defer mr.Close() - workers, err := s.ListWorkers(ctx) + workers, _, err := s.ListWorkers(ctx, 1000, "") if err != nil { t.Fatalf("ListWorkers failed: %v", err) } @@ -631,6 +631,89 @@ func TestListWorkers_Empty(t *testing.T) { } } +func TestListWorkers_Pagination(t *testing.T) { + mr, s, ctx := setupTest(t) + defer mr.Close() + + for i := 0; i < 5; i++ { + worker := &ateapipb.Worker{ + WorkerNamespace: "ns1", + WorkerPool: "pool1", + WorkerPod: fmt.Sprintf("pod%d", i), + } + if err := s.CreateWorker(ctx, worker); err != nil { + t.Fatalf("failed to create worker %d: %v", i, err) + } + } + + var allWorkers []*ateapipb.Worker + pageToken := "" + + for { + workers, nextToken, err := s.ListWorkers(ctx, 2, pageToken) + if err != nil { + t.Fatalf("ListWorkers failed: %v", err) + } + + allWorkers = append(allWorkers, workers...) + pageToken = nextToken + if pageToken == "" { + break + } + } + + if len(allWorkers) != 5 { + t.Fatalf("expected 5 workers total, got %d", len(allWorkers)) + } + + seen := make(map[string]bool) + for _, w := range allWorkers { + if seen[w.GetWorkerPod()] { + t.Errorf("duplicate worker found in paginated results: %s", w.GetWorkerPod()) + } + seen[w.GetWorkerPod()] = true + } +} + +func TestListAtespaces_Pagination(t *testing.T) { + mr, s, ctx := setupTest(t) + defer mr.Close() + + for i := 0; i < 5; i++ { + if _, err := s.CreateAtespace(ctx, newTestAtespace(fmt.Sprintf("team-%d", i))); err != nil { + t.Fatalf("failed to create atespace %d: %v", i, err) + } + } + + var allAtespaces []*ateapipb.Atespace + pageToken := "" + + for { + atespaces, nextToken, err := s.ListAtespaces(ctx, 2, pageToken) + if err != nil { + t.Fatalf("ListAtespaces failed: %v", err) + } + + allAtespaces = append(allAtespaces, atespaces...) + pageToken = nextToken + if pageToken == "" { + break + } + } + + if len(allAtespaces) != 5 { + t.Fatalf("expected 5 atespaces total, got %d", len(allAtespaces)) + } + + seen := make(map[string]bool) + for _, a := range allAtespaces { + if seen[a.GetMetadata().GetName()] { + t.Errorf("duplicate atespace found in paginated results: %s", a.GetMetadata().GetName()) + } + seen[a.GetMetadata().GetName()] = true + } +} + func TestListActors_Empty(t *testing.T) { mr, s, ctx := setupTest(t) defer mr.Close() @@ -1038,7 +1121,7 @@ func TestListAtespaces(t *testing.T) { t.Fatalf("CreateAtespace(%s) failed: %v", n, err) } } - got, err := s.ListAtespaces(ctx) + got, _, err := s.ListAtespaces(ctx, 1000, "") if err != nil { t.Fatalf("ListAtespaces failed: %v", err) } @@ -1060,7 +1143,7 @@ func TestListAtespaces_Empty(t *testing.T) { mr, s, ctx := setupTest(t) defer mr.Close() - got, err := s.ListAtespaces(ctx) + got, _, err := s.ListAtespaces(ctx, 1000, "") if err != nil { t.Fatalf("ListAtespaces failed: %v", err) } @@ -1316,3 +1399,132 @@ func TestListActors_MultiMaster_Pagination(t *testing.T) { seen[a.GetMetadata().GetName()] = true } } + +// newMultiMasterStore returns a Persistence whose master iteration spans +// numShards independent miniredis instances, plus a per-shard Persistence for +// seeding data onto a specific shard. +func newMultiMasterStore(t *testing.T, numShards int) (*Persistence, []*Persistence) { + t.Helper() + type shard struct { + client *redis.Client + clusterClient *redis.ClusterClient + } + var shards []shard + for i := 0; i < numShards; i++ { + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("failed to start miniredis %d: %v", i, err) + } + t.Cleanup(mr.Close) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + clusterClient := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{mr.Addr()}}) + t.Cleanup(func() { client.Close() }) + t.Cleanup(func() { clusterClient.Close() }) + shards = append(shards, shard{client: client, clusterClient: clusterClient}) + } + sort.Slice(shards, func(i, j int) bool { + return shards[i].client.Options().Addr < shards[j].client.Options().Addr + }) + var clients []*redis.Client + var perShard []*Persistence + for _, sh := range shards { + clients = append(clients, sh.client) + perShard = append(perShard, &Persistence{rdb: sh.clusterClient}) + } + fake := &concurrentMasterClient{redisClient: shards[0].clusterClient, masters: clients} + return &Persistence{rdb: fake}, perShard +} + +// TestListWorkers_MultiMaster_Pagination mirrors +// TestListActors_MultiMaster_Pagination for ListWorkers, sweeping page sizes +// so page boundaries both do and do not align with shard boundaries (the +// aligned case is the #425 shard-skip regression). +func TestListWorkers_MultiMaster_Pagination(t *testing.T) { + ctx := context.Background() + const numShards = 3 + for _, pageSize := range []int32{1, 2, 3, 4} { + t.Run(fmt.Sprintf("pageSize=%d", pageSize), func(t *testing.T) { + s, perShard := newMultiMasterStore(t, numShards) + for shardIdx, ps := range perShard { + for itemIdx := 0; itemIdx < 3; itemIdx++ { + worker := &ateapipb.Worker{ + WorkerNamespace: "ns", + WorkerPool: "pool", + WorkerPod: fmt.Sprintf("pod-shard%d-item%d", shardIdx, itemIdx), + } + if err := ps.CreateWorker(ctx, worker); err != nil { + t.Fatalf("failed to seed worker: %v", err) + } + } + } + + seen := make(map[string]bool) + pageToken := "" + for { + workers, next, err := s.ListWorkers(ctx, pageSize, pageToken) + if err != nil { + t.Fatalf("ListWorkers: %v", err) + } + for _, w := range workers { + if seen[w.GetWorkerPod()] { + t.Errorf("duplicate worker in paginated results: %s", w.GetWorkerPod()) + } + seen[w.GetWorkerPod()] = true + } + if next == "" { + break + } + pageToken = next + } + if len(seen) != numShards*3 { + t.Fatalf("expected %d workers across %d shards, got %d", numShards*3, numShards, len(seen)) + } + }) + } +} + +// TestListAtespaces_MultiMaster_Pagination mirrors +// TestListWorkers_MultiMaster_Pagination for ListAtespaces. +func TestListAtespaces_MultiMaster_Pagination(t *testing.T) { + ctx := context.Background() + const numShards = 3 + for _, pageSize := range []int32{1, 2, 3, 4} { + t.Run(fmt.Sprintf("pageSize=%d", pageSize), func(t *testing.T) { + s, perShard := newMultiMasterStore(t, numShards) + for shardIdx, ps := range perShard { + for itemIdx := 0; itemIdx < 3; itemIdx++ { + atespace := &ateapipb.Atespace{ + Metadata: &ateapipb.ResourceMetadata{ + Name: fmt.Sprintf("space-shard%d-item%d", shardIdx, itemIdx), + }, + } + if _, err := ps.CreateAtespace(ctx, atespace); err != nil { + t.Fatalf("failed to seed atespace: %v", err) + } + } + } + + seen := make(map[string]bool) + pageToken := "" + for { + atespaces, next, err := s.ListAtespaces(ctx, pageSize, pageToken) + if err != nil { + t.Fatalf("ListAtespaces: %v", err) + } + for _, a := range atespaces { + if seen[a.GetMetadata().GetName()] { + t.Errorf("duplicate atespace in paginated results: %s", a.GetMetadata().GetName()) + } + seen[a.GetMetadata().GetName()] = true + } + if next == "" { + break + } + pageToken = next + } + if len(seen) != numShards*3 { + t.Fatalf("expected %d atespaces across %d shards, got %d", numShards*3, numShards, len(seen)) + } + }) + } +} diff --git a/cmd/ateapi/internal/store/store.go b/cmd/ateapi/internal/store/store.go index aff2027c9..211dbd02c 100644 --- a/cmd/ateapi/internal/store/store.go +++ b/cmd/ateapi/internal/store/store.go @@ -68,8 +68,8 @@ type Interface interface { // Fetches an atespace by name. Returns ErrNotFound if missing. GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error) - // Lists all atespaces. Returns nil if none found. - ListAtespaces(ctx context.Context) ([]*ateapipb.Atespace, error) + // Lists atespaces. Returns a page of atespaces and a next page token. + ListAtespaces(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Atespace, string, error) // AtespaceExists reports whether the atespace object exists. AtespaceExists(ctx context.Context, name string) (bool, error) @@ -91,8 +91,8 @@ type Interface interface { // Removes a worker. Idempotent: does nothing if worker is not found. DeleteWorker(ctx context.Context, namespace, pool, pod string) error - // Lists all known workers. Returns nil if none found. - ListWorkers(ctx context.Context) ([]*ateapipb.Worker, error) + // Lists workers. Returns a page of workers and a next page token. + ListWorkers(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) // WatchWorkers returns an active subscription to track worker state changes. // The watch's Events channel is closed when the caller calls Close, the diff --git a/cmd/ateapi/internal/workercache/workercache.go b/cmd/ateapi/internal/workercache/workercache.go index f945faa6c..81b176573 100644 --- a/cmd/ateapi/internal/workercache/workercache.go +++ b/cmd/ateapi/internal/workercache/workercache.go @@ -31,6 +31,9 @@ import ( "k8s.io/apimachinery/pkg/util/wait" ) +// relistPageSize is the page size used for the relist. +const relistPageSize = 1000 + // Cache maintains an in-memory snapshot of all workers. // // TODO: add metrics — at minimum a gauge for worker count, a counter for @@ -99,9 +102,18 @@ func (c *Cache) sync(ctx context.Context) (*store.WorkerWatch, error) { } func (c *Cache) relist(ctx context.Context) error { - workers, err := c.store.ListWorkers(ctx) - if err != nil { - return fmt.Errorf("ListWorkers: %w", err) + var workers []*ateapipb.Worker + pageToken := "" + for { + page, nextToken, err := c.store.ListWorkers(ctx, relistPageSize, pageToken) + if err != nil { + return fmt.Errorf("ListWorkers: %w", err) + } + workers = append(workers, page...) + if nextToken == "" { + break + } + pageToken = nextToken } newMap := make(map[string]*ateapipb.Worker, len(workers)) for _, w := range workers { diff --git a/cmd/ateapi/internal/workercache/workercache_test.go b/cmd/ateapi/internal/workercache/workercache_test.go index ef6f7d322..cda1b8aea 100644 --- a/cmd/ateapi/internal/workercache/workercache_test.go +++ b/cmd/ateapi/internal/workercache/workercache_test.go @@ -380,15 +380,15 @@ func (f *fakeStore) WatchWorkers(_ context.Context) (*store.WorkerWatch, error) }), nil } -func (f *fakeStore) ListWorkers(_ context.Context) ([]*ateapipb.Worker, error) { +func (f *fakeStore) ListWorkers(_ context.Context, _ int32, _ string) ([]*ateapipb.Worker, string, error) { f.mu.Lock() defer f.mu.Unlock() if f.listErr != nil { - return nil, f.listErr + return nil, "", f.listErr } out := make([]*ateapipb.Worker, len(f.workers)) copy(out, f.workers) - return out, nil + return out, "", nil } func (f *fakeStore) send(event store.WorkerEvent) { diff --git a/cmd/kubectl-ate/internal/cmd/get_atespaces.go b/cmd/kubectl-ate/internal/cmd/get_atespaces.go index 9c456c253..e63eccc79 100644 --- a/cmd/kubectl-ate/internal/cmd/get_atespaces.go +++ b/cmd/kubectl-ate/internal/cmd/get_atespaces.go @@ -43,11 +43,24 @@ var getAtespacesCmd = &cobra.Command{ return printer.PrintAtespace(resp, outputFmt) } - resp, err := apiClient.ListAtespaces(ctx, &ateapipb.ListAtespacesRequest{}) - if err != nil { - return fmt.Errorf("failed to list atespaces: %w", err) + var allAtespaces []*ateapipb.Atespace + pageToken := "" + for { + resp, err := apiClient.ListAtespaces(ctx, &ateapipb.ListAtespacesRequest{ + PageSize: 1000, + PageToken: pageToken, + }) + if err != nil { + return fmt.Errorf("failed to list atespaces: %w", err) + } + allAtespaces = append(allAtespaces, resp.GetAtespaces()...) + + pageToken = resp.GetNextPageToken() + if pageToken == "" { + break + } } - return printer.PrintAtespaces(resp.GetAtespaces(), outputFmt) + return printer.PrintAtespaces(allAtespaces, outputFmt) }, } diff --git a/cmd/kubectl-ate/internal/cmd/get_workers.go b/cmd/kubectl-ate/internal/cmd/get_workers.go index 2d9155680..7d0fa61a2 100644 --- a/cmd/kubectl-ate/internal/cmd/get_workers.go +++ b/cmd/kubectl-ate/internal/cmd/get_workers.go @@ -35,11 +35,24 @@ var getWorkersCmd = &cobra.Command{ } defer apiClient.Close() - resp, err := apiClient.ListWorkers(ctx, &ateapipb.ListWorkersRequest{}) - if err != nil { - return fmt.Errorf("failed to list workers: %w", err) + var allWorkers []*ateapipb.Worker + pageToken := "" + for { + resp, err := apiClient.ListWorkers(ctx, &ateapipb.ListWorkersRequest{ + PageSize: 1000, + PageToken: pageToken, + }) + if err != nil { + return fmt.Errorf("failed to list workers: %w", err) + } + allWorkers = append(allWorkers, resp.GetWorkers()...) + + pageToken = resp.GetNextPageToken() + if pageToken == "" { + break + } } - return printer.PrintWorkers(resp.GetWorkers(), outputFmt) + return printer.PrintWorkers(allWorkers, outputFmt) }, } diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 0350f9fd9..c022a73d7 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -757,7 +757,14 @@ func (x *GetAtespaceRequest) GetAtespace() *ObjectRef { } type ListAtespacesRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` + state protoimpl.MessageState `protogen:"open.v1"` + // Requested page size; the server may return fewer, or occasionally + // slightly more. If unspecified, defaults to a server-chosen value; + // values above 1000 are coerced to 1000. + PageSize int32 `protobuf:"varint,1,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` + // Pagination token from a previous ListAtespaces response. + // Omit or leave empty for the first request. + PageToken string `protobuf:"bytes,2,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -792,9 +799,26 @@ func (*ListAtespacesRequest) Descriptor() ([]byte, []int) { return file_ateapi_proto_rawDescGZIP(), []int{10} } +func (x *ListAtespacesRequest) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 +} + +func (x *ListAtespacesRequest) GetPageToken() string { + if x != nil { + return x.PageToken + } + return "" +} + type ListAtespacesResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Atespaces []*Atespace `protobuf:"bytes,1,rep,name=atespaces,proto3" json:"atespaces,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + // The page of atespaces. This list may be empty even if there are more results. + Atespaces []*Atespace `protobuf:"bytes,1,rep,name=atespaces,proto3" json:"atespaces,omitempty"` + // Pagination token for the next page. Empty if this is the last page. + NextPageToken string `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -836,6 +860,13 @@ func (x *ListAtespacesResponse) GetAtespaces() []*Atespace { return nil } +func (x *ListAtespacesResponse) GetNextPageToken() string { + if x != nil { + return x.NextPageToken + } + return "" +} + type DeleteAtespaceRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Atespace *ObjectRef `protobuf:"bytes,1,opt,name=atespace,proto3" json:"atespace,omitempty"` @@ -1389,7 +1420,14 @@ func (x *DeleteActorRequest) GetActor() *ObjectRef { } type ListWorkersRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` + state protoimpl.MessageState `protogen:"open.v1"` + // Requested page size; the server may return fewer, or occasionally + // slightly more. If unspecified, defaults to a server-chosen value; + // values above 1000 are coerced to 1000. + PageSize int32 `protobuf:"varint,1,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` + // Pagination token from a previous ListWorkers response. + // Omit or leave empty for the first request. + PageToken string `protobuf:"bytes,2,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1424,9 +1462,26 @@ func (*ListWorkersRequest) Descriptor() ([]byte, []int) { return file_ateapi_proto_rawDescGZIP(), []int{24} } +func (x *ListWorkersRequest) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 +} + +func (x *ListWorkersRequest) GetPageToken() string { + if x != nil { + return x.PageToken + } + return "" +} + type ListWorkersResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Workers []*Worker `protobuf:"bytes,1,rep,name=workers,proto3" json:"workers,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + // The page of workers. This list may be empty even if there are more results. + Workers []*Worker `protobuf:"bytes,1,rep,name=workers,proto3" json:"workers,omitempty"` + // Pagination token for the next page. Empty if this is the last page. + NextPageToken string `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1468,19 +1523,26 @@ func (x *ListWorkersResponse) GetWorkers() []*Worker { return nil } -// Request to list actors with pagination. -// This operation provides "soft" guarantees: actors may be missed or duplicated -// if the system state changes during pagination. +func (x *ListWorkersResponse) GetNextPageToken() string { + if x != nil { + return x.NextPageToken + } + return "" +} + +// List actors with pagination. Provides "soft" guarantees: actors may be +// missed or duplicated if the system state changes during pagination. type ListActorsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // The maximum number of actors to return. The server may enforce a hard limit - // (e.g., 1000) regardless of the requested size. - PageSize int32 `protobuf:"varint,1,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` - // An opaque pagination token obtained from a previous ListActorsResponse. - // Empty for the first request. - PageToken string `protobuf:"bytes,2,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` - // If set, list only actors in this atespace (scoped SCAN actor::*). - Atespace string `protobuf:"bytes,3,opt,name=atespace,proto3" json:"atespace,omitempty"` + // The atespace to list actors from. Empty lists across all atespaces. + Atespace string `protobuf:"bytes,1,opt,name=atespace,proto3" json:"atespace,omitempty"` + // Requested page size; the server may return fewer, or occasionally + // slightly more. If unspecified, defaults to a server-chosen value; + // values above 1000 are coerced to 1000. + PageSize int32 `protobuf:"varint,2,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` + // Pagination token from a previous ListActors response. + // Omit or leave empty for the first request. + PageToken string `protobuf:"bytes,3,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1515,6 +1577,13 @@ func (*ListActorsRequest) Descriptor() ([]byte, []int) { return file_ateapi_proto_rawDescGZIP(), []int{26} } +func (x *ListActorsRequest) GetAtespace() string { + if x != nil { + return x.Atespace + } + return "" +} + func (x *ListActorsRequest) GetPageSize() int32 { if x != nil { return x.PageSize @@ -1529,20 +1598,11 @@ func (x *ListActorsRequest) GetPageToken() string { return "" } -func (x *ListActorsRequest) GetAtespace() string { - if x != nil { - return x.Atespace - } - return "" -} - -// Response for a ListActors operation. type ListActorsResponse struct { state protoimpl.MessageState `protogen:"open.v1"` // The page of actors. This list may be empty even if there are more results. Actors []*Actor `protobuf:"bytes,1,rep,name=actors,proto3" json:"actors,omitempty"` - // An opaque token to retrieve the next page of results. - // If empty, there are no more results. Clients must loop until this is empty. + // Pagination token for the next page. Empty if this is the last page. NextPageToken string `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -2194,10 +2254,14 @@ const file_ateapi_proto_rawDesc = "" + "\x15CreateAtespaceRequest\x12,\n" + "\batespace\x18\x01 \x01(\v2\x10.ateapi.AtespaceR\batespace\"C\n" + "\x12GetAtespaceRequest\x12-\n" + - "\batespace\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\batespace\"\x16\n" + - "\x14ListAtespacesRequest\"G\n" + + "\batespace\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\batespace\"R\n" + + "\x14ListAtespacesRequest\x12\x1b\n" + + "\tpage_size\x18\x01 \x01(\x05R\bpageSize\x12\x1d\n" + + "\n" + + "page_token\x18\x02 \x01(\tR\tpageToken\"o\n" + "\x15ListAtespacesResponse\x12.\n" + - "\tatespaces\x18\x01 \x03(\v2\x10.ateapi.AtespaceR\tatespaces\"F\n" + + "\tatespaces\x18\x01 \x03(\v2\x10.ateapi.AtespaceR\tatespaces\x12&\n" + + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"F\n" + "\x15DeleteAtespaceRequest\x12-\n" + "\batespace\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\batespace\":\n" + "\x0fGetActorRequest\x12'\n" + @@ -2223,15 +2287,19 @@ const file_ateapi_proto_rawDesc = "" + "\x13ResumeActorResponse\x12#\n" + "\x05actor\x18\x01 \x01(\v2\r.ateapi.ActorR\x05actor\"=\n" + "\x12DeleteActorRequest\x12'\n" + - "\x05actor\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\x05actor\"\x14\n" + - "\x12ListWorkersRequest\"?\n" + - "\x13ListWorkersResponse\x12(\n" + - "\aworkers\x18\x01 \x03(\v2\x0e.ateapi.WorkerR\aworkers\"k\n" + - "\x11ListActorsRequest\x12\x1b\n" + + "\x05actor\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\x05actor\"P\n" + + "\x12ListWorkersRequest\x12\x1b\n" + "\tpage_size\x18\x01 \x01(\x05R\bpageSize\x12\x1d\n" + "\n" + - "page_token\x18\x02 \x01(\tR\tpageToken\x12\x1a\n" + - "\batespace\x18\x03 \x01(\tR\batespace\"c\n" + + "page_token\x18\x02 \x01(\tR\tpageToken\"g\n" + + "\x13ListWorkersResponse\x12(\n" + + "\aworkers\x18\x01 \x03(\v2\x0e.ateapi.WorkerR\aworkers\x12&\n" + + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"k\n" + + "\x11ListActorsRequest\x12\x1a\n" + + "\batespace\x18\x01 \x01(\tR\batespace\x12\x1b\n" + + "\tpage_size\x18\x02 \x01(\x05R\bpageSize\x12\x1d\n" + + "\n" + + "page_token\x18\x03 \x01(\tR\tpageToken\"c\n" + "\x12ListActorsResponse\x12%\n" + "\x06actors\x18\x01 \x03(\v2\r.ateapi.ActorR\x06actors\x12&\n" + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"\xa8\x03\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 4c1d0461f..5fe2d194d 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -45,10 +45,10 @@ service Control { // Delete an actor. Only suspended actors can be deleted. rpc DeleteActor(DeleteActorRequest) returns (Actor) {} - // List all workers currently reflected in redis. + // List Workers. rpc ListWorkers(ListWorkersRequest) returns (ListWorkersResponse) {} - // List all actors currently reflected in redis. + // List Actors. rpc ListActors(ListActorsRequest) returns (ListActorsResponse) {} // Create a new Atespace. Substrate-native, stored in Redis. @@ -57,7 +57,7 @@ service Control { // Get an Atespace by name. rpc GetAtespace(GetAtespaceRequest) returns (Atespace) {} - // List all Atespaces. + // List Atespaces. rpc ListAtespaces(ListAtespacesRequest) returns (ListAtespacesResponse) {} // Delete an empty Atespace. Rejects (FailedPrecondition) if any actors remain. @@ -184,9 +184,23 @@ message GetAtespaceRequest { ObjectRef atespace = 1; } -message ListAtespacesRequest {} +message ListAtespacesRequest { + // Requested page size; the server may return fewer, or occasionally + // slightly more. If unspecified, defaults to a server-chosen value; + // values above 1000 are coerced to 1000. + int32 page_size = 1; + + // Pagination token from a previous ListAtespaces response. + // Omit or leave empty for the first request. + string page_token = 2; +} + message ListAtespacesResponse { + // The page of atespaces. This list may be empty even if there are more results. repeated Atespace atespaces = 1; + + // Pagination token for the next page. Empty if this is the last page. + string next_page_token = 2; } message DeleteAtespaceRequest { @@ -249,35 +263,46 @@ message DeleteActorRequest { ObjectRef actor = 1; } -message ListWorkersRequest {} +message ListWorkersRequest { + // Requested page size; the server may return fewer, or occasionally + // slightly more. If unspecified, defaults to a server-chosen value; + // values above 1000 are coerced to 1000. + int32 page_size = 1; + + // Pagination token from a previous ListWorkers response. + // Omit or leave empty for the first request. + string page_token = 2; +} message ListWorkersResponse { + // The page of workers. This list may be empty even if there are more results. repeated Worker workers = 1; + + // Pagination token for the next page. Empty if this is the last page. + string next_page_token = 2; } -// Request to list actors with pagination. -// This operation provides "soft" guarantees: actors may be missed or duplicated -// if the system state changes during pagination. +// List actors with pagination. Provides "soft" guarantees: actors may be +// missed or duplicated if the system state changes during pagination. message ListActorsRequest { - // The maximum number of actors to return. The server may enforce a hard limit - // (e.g., 1000) regardless of the requested size. - int32 page_size = 1; + // The atespace to list actors from. Empty lists across all atespaces. + string atespace = 1; - // An opaque pagination token obtained from a previous ListActorsResponse. - // Empty for the first request. - string page_token = 2; + // Requested page size; the server may return fewer, or occasionally + // slightly more. If unspecified, defaults to a server-chosen value; + // values above 1000 are coerced to 1000. + int32 page_size = 2; - // If set, list only actors in this atespace (scoped SCAN actor::*). - string atespace = 3; + // Pagination token from a previous ListActors response. + // Omit or leave empty for the first request. + string page_token = 3; } -// Response for a ListActors operation. message ListActorsResponse { // The page of actors. This list may be empty even if there are more results. repeated Actor actors = 1; - // An opaque token to retrieve the next page of results. - // If empty, there are no more results. Clients must loop until this is empty. + // Pagination token for the next page. Empty if this is the last page. string next_page_token = 2; } diff --git a/pkg/proto/ateapipb/ateapi_grpc.pb.go b/pkg/proto/ateapipb/ateapi_grpc.pb.go index 59eaeffc2..7b8fb44f4 100644 --- a/pkg/proto/ateapipb/ateapi_grpc.pb.go +++ b/pkg/proto/ateapipb/ateapi_grpc.pb.go @@ -71,15 +71,15 @@ type ControlClient interface { ResumeActor(ctx context.Context, in *ResumeActorRequest, opts ...grpc.CallOption) (*ResumeActorResponse, error) // Delete an actor. Only suspended actors can be deleted. DeleteActor(ctx context.Context, in *DeleteActorRequest, opts ...grpc.CallOption) (*Actor, error) - // List all workers currently reflected in redis. + // List Workers. ListWorkers(ctx context.Context, in *ListWorkersRequest, opts ...grpc.CallOption) (*ListWorkersResponse, error) - // List all actors currently reflected in redis. + // List Actors. ListActors(ctx context.Context, in *ListActorsRequest, opts ...grpc.CallOption) (*ListActorsResponse, error) // Create a new Atespace. Substrate-native, stored in Redis. CreateAtespace(ctx context.Context, in *CreateAtespaceRequest, opts ...grpc.CallOption) (*Atespace, error) // Get an Atespace by name. GetAtespace(ctx context.Context, in *GetAtespaceRequest, opts ...grpc.CallOption) (*Atespace, error) - // List all Atespaces. + // List Atespaces. ListAtespaces(ctx context.Context, in *ListAtespacesRequest, opts ...grpc.CallOption) (*ListAtespacesResponse, error) // Delete an empty Atespace. Rejects (FailedPrecondition) if any actors remain. DeleteAtespace(ctx context.Context, in *DeleteAtespaceRequest, opts ...grpc.CallOption) (*Atespace, error) @@ -255,15 +255,15 @@ type ControlServer interface { ResumeActor(context.Context, *ResumeActorRequest) (*ResumeActorResponse, error) // Delete an actor. Only suspended actors can be deleted. DeleteActor(context.Context, *DeleteActorRequest) (*Actor, error) - // List all workers currently reflected in redis. + // List Workers. ListWorkers(context.Context, *ListWorkersRequest) (*ListWorkersResponse, error) - // List all actors currently reflected in redis. + // List Actors. ListActors(context.Context, *ListActorsRequest) (*ListActorsResponse, error) // Create a new Atespace. Substrate-native, stored in Redis. CreateAtespace(context.Context, *CreateAtespaceRequest) (*Atespace, error) // Get an Atespace by name. GetAtespace(context.Context, *GetAtespaceRequest) (*Atespace, error) - // List all Atespaces. + // List Atespaces. ListAtespaces(context.Context, *ListAtespacesRequest) (*ListAtespacesResponse, error) // Delete an empty Atespace. Rejects (FailedPrecondition) if any actors remain. DeleteAtespace(context.Context, *DeleteAtespaceRequest) (*Atespace, error)