Fix: Only run worker id action getter query when needed (#3976)

* fix: add if block to check if any ids / hashes are present

* fix: add some tracing

* feat: add pagination

* feat: server-side status filter

* fix: conditions

* chore: lint

* fix: backwards compat

* fix: backwards compat

* chore: couple more

Author: mrkaye97

api-contracts/openapi/components/schemas/_index.yaml

@@ -252,6 +252,8 @@ WorkerRuntimeInfo:
   $ref: "./worker.yaml#/WorkerRuntimeInfo"
 WorkerRuntimeSDKs:
   $ref: "./worker.yaml#/WorkerRuntimeSDKs"
+WorkerStatus:
+  $ref: "./worker.yaml#/WorkerStatus"
 WorkerList:
   $ref: "./worker.yaml#/WorkerList"
 SemaphoreSlots:

api-contracts/openapi/components/schemas/worker.yaml

@@ -143,12 +143,8 @@ Worker:
       items:
         $ref: "./_index.yaml#/RecentStepRuns"
     status:
-      type: string
+      $ref: "#/WorkerStatus"
       description: The status of the worker.
-      enum:
-        - ACTIVE
-        - INACTIVE
-        - PAUSED
     slotConfig:
       type: object
       description: Slot availability and limits for this worker (slot_type -> { available, limit }).
@@ -217,3 +213,10 @@ WorkerRuntimeInfo:
       type: string
     runtimeExtra:
       type: string
+
+WorkerStatus:
+  type: string
+  enum:
+    - ACTIVE
+    - INACTIVE
+    - PAUSED

api-contracts/openapi/paths/worker/worker.yaml

@@ -13,6 +13,28 @@ withTenant:
           format: uuid
           minLength: 36
           maxLength: 36
+      - description: The number to skip
+        in: query
+        name: offset
+        required: false
+        schema:
+          type: integer
+          format: int64
+      - description: The number to limit by
+        in: query
+        name: limit
+        required: false
+        schema:
+          type: integer
+          format: int64
+      - description: Filter by worker status
+        in: query
+        name: statuses
+        required: false
+        schema:
+          type: array
+          items:
+            $ref: "../../components/schemas/_index.yaml#/WorkerStatus"
     responses:
       "200":
         content:

api/v1/server/handlers/workers/list.go

@@ -2,6 +2,7 @@ package workers
 
 import (
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/google/uuid"
@@ -35,8 +36,31 @@ func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, re
 
 	sixSecAgo := time.Now().Add(-24 * time.Hour)
 
+	limit := 50
+	offset := 0
+
 	opts := &v1.ListWorkersOpts{
 		LastHeartbeatAfter: &sixSecAgo,
+		Limit:              &limit,
+		Offset:             &offset,
+	}
+
+	if request.Params.Limit != nil {
+		limit = int(*request.Params.Limit)
+		opts.Limit = &limit
+	}
+
+	if request.Params.Offset != nil {
+		offset = int(*request.Params.Offset)
+		opts.Offset = &offset
+	}
+
+	if request.Params.Statuses != nil {
+		statuses := make([]string, len(*request.Params.Statuses))
+		for i, s := range *request.Params.Statuses {
+			statuses[i] = string(s)
+		}
+		opts.Statuses = statuses
 	}
 
 	_, listSpan := telemetry.NewSpan(reqCtx, "worker-service.v0.list-workers")
@@ -46,7 +70,7 @@ func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		telemetry.AttributeKV{Key: "tenant.id", Value: tenant.ID},
 	)
 
-	workers, err := t.config.V1.Workers().ListWorkers(reqCtx, tenantId, opts)
+	workers, count, err := t.config.V1.Workers().ListWorkers(reqCtx, tenantId, opts)
 
 	if err != nil {
 		listSpan.RecordError(err)
@@ -75,9 +99,22 @@ func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		rows[i] = *transformers.ToWorkerSqlc(&workerCp.Worker, slotConfig, nil)
 	}
 
+	totalPages := int64(math.Ceil(float64(count) / float64(limit)))
+	currPage := 1 + int64(math.Ceil(float64(offset)/float64(limit)))
+	nextPage := currPage + 1
+
+	if currPage == totalPages {
+		nextPage = currPage
+	}
+
 	return gen.WorkerList200JSONResponse(
 		gen.WorkerList{
 			Rows: &rows,
+			Pagination: &gen.PaginationResponse{
+				NumPages:    &totalPages,
+				CurrentPage: &currPage,
+				NextPage:    &nextPage,
+			},
 		},
 	), nil
 }
@@ -88,8 +125,31 @@ func (t *WorkerService) workerListV1(ctx echo.Context, tenant *sqlcv1.Tenant, re
 
 	sixSecAgo := time.Now().Add(-24 * time.Hour)
 
+	limit := 50
+	offset := 0
+
 	opts := &v1.ListWorkersOpts{
 		LastHeartbeatAfter: &sixSecAgo,
+		Limit:              &limit,
+		Offset:             &offset,
+	}
+
+	if request.Params.Limit != nil {
+		limit = int(*request.Params.Limit)
+		opts.Limit = &limit
+	}
+
+	if request.Params.Offset != nil {
+		offset = int(*request.Params.Offset)
+		opts.Offset = &offset
+	}
+
+	if request.Params.Statuses != nil {
+		statuses := make([]string, len(*request.Params.Statuses))
+		for i, s := range *request.Params.Statuses {
+			statuses[i] = string(s)
+		}
+		opts.Statuses = statuses
 	}
 
 	listCtx, listSpan := telemetry.NewSpan(reqCtx, "worker-service.v1.list-workers")
@@ -99,7 +159,7 @@ func (t *WorkerService) workerListV1(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		telemetry.AttributeKV{Key: "tenant.id", Value: tenant.ID},
 	)
 
-	workerRows, err := t.config.V1.Workers().ListWorkers(listCtx, tenantId, opts)
+	workerRows, count, err := t.config.V1.Workers().ListWorkers(listCtx, tenantId, opts)
 
 	if err != nil {
 		listSpan.RecordError(err)
@@ -173,9 +233,22 @@ func (t *WorkerService) workerListV1(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		rows[i] = *transformersv1.ToWorkerSqlc(&workerCp, slotConfig, actions, nil, labels)
 	}
 
+	totalPages := int64(math.Ceil(float64(count) / float64(limit)))
+	currPage := 1 + int64(math.Ceil(float64(offset)/float64(limit)))
+	nextPage := currPage + 1
+
+	if currPage == totalPages {
+		nextPage = currPage
+	}
+
 	return gen.WorkerList200JSONResponse(
 		gen.WorkerList{
 			Rows: &rows,
+			Pagination: &gen.PaginationResponse{
+				NumPages:    &totalPages,
+				CurrentPage: &currPage,
+				NextPage:    &nextPage,
+			},
 		},
 	), nil
 }