mirror of
https://github.com/agent-substrate/substrate.git
synced 2026-11-11 10:23:17 +08:00
Grant McCloskey
06dbd13ace
# Implement Pagination and Pipelining for ListActors ## Overview **Fixes:** #153 This PR introduces robust, memory-safe pagination to the `ListActors` operation and optimizes the Valkey retrieval layer. Previously, `ListActors` performed an unbounded parallel scatter-gather across all Valkey cluster shards, pulling all actors into memory at once. At scale (millions/billions of actors), this would cause OOM errors and severe latency. This update replaces the concurrent scatter-gather with a deterministic, sequential shard scanning approach, adhering to [AIP-158](https://google.aip.dev/158) pagination standards. ## Changes Included - **gRPC API (`ateapi.proto`)**: Added `page_size`, `page_token`, and `next_page_token` fields. - **Client (`kubectl-ate`)**: Updated to seamlessly loop through `next_page_token` values until all actors are fetched, presenting a continuous stream to the user without blowing up memory on a single request. - **API Handler (`controlapi`)**: Bounded requests with a max `page_size` limit of `1000`. - **Persistence (`ateredis`)**: - Implemented `listActorsPageToken` to safely serialize and encode cross-shard cursors. - Replaced `ForEachMaster` concurrent execution with deterministic sequential iteration. - Implemented Valkey Pipeline batching (`Pipelined`) for `GET` requests to eliminate N+1 network latency. - **Testing**: Added `TestListActors_Pagination` at both the integration level (`functional_test.go`) and data layer (`ateredis_test.go`). Refactored sorting to use `cmpopts.SortSlices`. ## Design Decisions During the design phase, several explicit trade-offs were made to balance performance with API simplicity: 1. **Opaque Tokens (AIP-158):** The `page_token` is strictly an opaque string at the `ateapi` layer. The client (`kubectl-ate`) is completely unaware that it contains base64-encoded JSON mapping to `{"shard", "cursor"}`. This keeps our API database-agnostic. 2. **Empty Pages & Client-Side Looping:** The persistence layer is kept simple and "dumb". If a shard scan returns 0 keys but the cursor is not `0`, we return an empty page. We rely on the client `kubectl-ate` to `for` loop until `next_page_token` is completely exhausted. Update: I have server side looping across shards until pagesize is met to limit round trips due to high shard count or low actor count on distributed shards. 3. **"Soft" Consistency (Missing/Duplicate Data):** `ListActors` is an infrequent operations command, not on the critical path. Taking a global cluster lock for iteration would severely degrade performance. We accept eventual consistency here: if actors migrate shards during an active scan, they might be duplicated or missed. 4. **Topology Flux Errors:** If a client passes a valid token pointing to a shard address that no longer exists (due to a cluster resharding mid-scan), the API simply returns an `Aborted` error. We do not attempt complex recovery; the client must just restart the `ListActors` command. 5. **Bounded Page Size:** Requests are currently capped at a maximum of `1000` to prevent single massive payloads. We can revisit/change the max later. NOTE: Actor filtering will be implemented later. ## Architecture Flow (Valkey) Note: If there are future plugin-db models they will need to handle the cusor/pages at their own implementation layer per the store.go interface. The system now seamlessly iterates across shards, handing off cursors safely between nodes: ```mermaid sequenceDiagram participant CLI as kubectl-ate participant API as ateapi participant Valkey as Valkey Cluster (Multiple Shards) CLI->>API: ListActors(page_size=1000, page_token="") Note over API: Request 1: Empty Token API->>API: Decode page_token (empty -> start fresh) API->>Valkey: Fetch all Master Nodes Valkey-->>API: [Master C, Master A, Master B] API->>API: Sort Masters alphabetically by IP/Port Note over API: Sorted: [Master A, Master B, Master C] API->>Valkey: SCAN (Master A, cursor=0, count=1000) Valkey-->>API: keys=[k1..k300], next_cursor=45 Note over API: OPTIMIZATION: Pipelining avoids N+1 problem API->>Valkey: PIPELINE GET (k1..k300) directly on Master A Valkey-->>API: values=[v1..v300] API->>API: Decode Protobufs API->>API: Encode page_token={"shard": "Master A", "cursor": 45} API-->>CLI: 300 Actors, next_page_token="base64(eyJzaGFyZ..." %% SECOND REQUEST CLI->>API: ListActors(page_size=1000, page_token="base64(...)") Note over API: Request 2: Resuming Mid-Shard API->>API: Decode page_token -> Shard="Master A", Cursor=45 API->>API: Fast-forward directly to Master A API->>Valkey: SCAN (Master A, cursor=45, count=1000) Valkey-->>API: keys=[k301..k400], next_cursor=0 API->>Valkey: PIPELINE GET (k301..k400) directly on Master A Valkey-->>API: values=[v301..v400] Note over API: Shard A is empty. Token points to Shard B! API->>API: Encode page_token={"shard": "Master B", "cursor": 0} API-->>CLI: 100 Actors, next_page_token="base64(eyJzaGFyZ..." %% THIRD REQUEST CLI->>API: ListActors(page_size=1000, page_token="base64(...)") Note over API: Request 3: Starting New Shard API->>API: Decode page_token -> Shard="Master B", Cursor=0 Note over API: Continues scanning Master B... ``` ## Testing: ``` > ./bin/kubectl-ate get actors NAMESPACE TEMPLATE ID STATUS ATEOM POD ATEOM IP VERSION ate-demo-counter counter 61b478cc-2e44-4ec7-ac6c-1af3f385de86 STATUS_SUSPENDED <none> 5 ``` ``` ~/go/bin/grpcurl -insecure -d '{"page_size": 2}' localhost:9090 ateapi.Control/ListActors { "actors": [ { "actorId": "test-counter-4", "version": "1", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED" } ], "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjE5OjYzNzkiLCJjdXJzb3IiOjI1MjU1fQ==" } ``` <details> ``` bash -c ' export NEXT_TOKEN="" while :; do echo "--- Fetching Page ---" # Fetch the data and store the response RESPONSE=$(~/go/bin/grpcurl -insecure -d "{\"page_size\": 2, \"page_token\": \"$NEXT_TOKEN\"}" localhost:9090 ateapi.Control/ListActors) # Print the response to the terminal echo "$RESPONSE" # Extract the token without needing jq NEXT_TOKEN=$(echo "$RESPONSE" | grep -o "\"nextPageToken\": \"[^\"]*\"" | cut -d"\"" -f4) # If no token is found, we reached the end if [ -z "$NEXT_TOKEN" ]; then echo "--- Reached the end! ---" break fi echo "" read -p "Press Enter to fetch the next page using the new token..." done ' ``` --- Fetching Page --- { "actors": [ { "actorId": "test-counter-4", "version": "1", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED" } ], "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjE5OjYzNzkiLCJjdXJzb3IiOjI1MjU1fQ==" } Press Enter to fetch the next page using the new token... --- Fetching Page --- { "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjIxOjYzNzkiLCJjdXJzb3IiOjB9" } Press Enter to fetch the next page using the new token... --- Fetching Page --- { "actors": [ { "actorId": "test-counter-1", "version": "1", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED" } ], "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjIxOjYzNzkiLCJjdXJzb3IiOjQ1NjAwfQ==" } Press Enter to fetch the next page using the new token... --- Fetching Page --- { "actors": [ { "actorId": "test-counter-5", "version": "1", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED" } ], "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjIxOjYzNzkiLCJjdXJzb3IiOjYyMDg2fQ==" } Press Enter to fetch the next page using the new token... --- Fetching Page --- { "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjI1OjYzNzkiLCJjdXJzb3IiOjB9" } Press Enter to fetch the next page using the new token... --- Fetching Page --- { "actors": [ { "actorId": "test-counter-2", "version": "1", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED" } ], "nextPageToken": "eyJzaGFyZCI6IjEwLjI0NC4wLjI1OjYzNzkiLCJjdXJzb3IiOjE2OTkzfQ==" } Press Enter to fetch the next page using the new token... --- Fetching Page --- { "actors": [ { "actorId": "test-counter-3", "version": "1", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED" }, { "actorId": "61b478cc-2e44-4ec7-ac6c-1af3f385de86", "version": "5", "actorTemplateNamespace": "ate-demo-counter", "actorTemplateName": "counter", "status": "STATUS_SUSPENDED", "lastSnapshot": "gs://ate-snapshots/ate-demo-counter/61b478cc-2e44-4ec7-ac6c-1af3f385de86/2026-06-03T03:17:34Z-UMYVUZOWW6LDPS5X5FSG3KC4VV" } ] } --- Reached the end! --- </details>