fix: close Lease and Schedule review follow-ups

public/openapi.yml

@@ -1965,6 +1965,12 @@ paths:
           required: false
           schema:
             type: string
+        - name: operation
+          in: query
+          required: false
+          description: Restrict pending handoffs to one operation before applying the response limit.
+          schema:
+            type: string
         - name: limit
           in: query
           required: false
@@ -2004,6 +2010,21 @@ paths:
         - $ref: "#/components/parameters/RealmPath"
         - $ref: "#/components/parameters/AreaPath"
         - $ref: "#/components/parameters/ResourcePath"
+        - name: operation
+          in: query
+          required: false
+          description: Restrict observations to one operation before applying the response limit.
+          schema:
+            type: string
+        - name: offset
+          in: query
+          required: false
+          description: Zero-based schedule-operation offset applied after resource and operation filtering.
+          schema:
+            type: integer
+            format: int64
+            default: 0
+            minimum: 0
         - name: limit
           in: query
           required: false
@@ -5509,6 +5530,7 @@ components:
           next_run,
           last_run,
           executions_total,
+          pending_handoffs,
         ]
       properties:
         route_family:
@@ -5529,18 +5551,23 @@ components:
         executions_total:
           type: integer
           format: int64
+        pending_handoffs:
+          type: integer
+          description: Current persisted handoff claims awaiting acknowledgement for this schedule operation.
 
     ScheduleExecutionObservationList:
       type: object
-      required: [route_family, realm, area, resource, limit, observations]
+      required: [route_family, realm, area, resource, offset, limit, has_more, observations]
       properties:
         route_family:
           type: integer
           format: int64
         realm: { type: string }
         area: { type: string }
         resource: { type: string }
+        offset: { type: integer }
         limit: { type: integer }
+        has_more: { type: boolean }
         observations:
           type: array
           items:

src/api/admin/handlers/routing/hierarchical_get.rs

@@ -97,20 +97,16 @@ fn handle_domain_collection_routes(
             };
             Some(list::stream_search(runtime.as_ref(), &request))
         }
-        ["missed"] if scheme == "schedule" => Some(list::schedule_missed_observations(
-            runtime.as_ref(),
-            match require_family_for_search(uri, principal, scope) {
-                Ok(family) => family,
-                Err(response) => return Some(*response),
-            },
-            list::parse_optional_string_query_param(uri, "realm").as_deref(),
-            list::parse_optional_string_query_param(uri, "area").as_deref(),
-            list::parse_optional_string_query_param(uri, "resource").as_deref(),
-            match parse_admin_record_limit(uri) {
-                Ok(limit) => limit,
+        ["missed"] if scheme == "schedule" => {
+            let request = match build_schedule_missed_observation_request(uri, principal, scope) {
+                Ok(request) => request,
                 Err(response) => return Some(*response),
-            },
-        )),
+            };
+            Some(list::schedule_missed_observations(
+                runtime.as_ref(),
+                &request,
+            ))
+        }
         ["search"] if scheme == "lease" => {
             let request = match build_lease_search_request(uri, principal, scope) {
                 Ok(request) => request,
@@ -552,14 +548,10 @@ fn handle_schedule_executions(
     area: &str,
     resource: &str,
 ) -> Response {
-    let family = match require_concrete_route_family(scope, uri, principal) {
-        Ok(family) => family,
+    let request = match build_schedule_execution_observation_request(uri, principal, scope) {
+        Ok(request) => request,
         Err(response) => return *response,
     };
-    let limit = match list::parse_admin_record_limit(uri) {
-        Ok(limit) => limit,
-        Err(message) => return error_response(StatusCode::BAD_REQUEST, &message),
-    };
 
     list::schedule_executions_for_resource(
         runtime.as_ref(),
@@ -568,8 +560,7 @@ fn handle_schedule_executions(
             area,
             resource,
         },
-        family,
-        limit,
+        &request,
     )
 }
 
@@ -832,3 +823,37 @@ fn build_rpc_call_observation_request(
         limit: parse_admin_record_limit(uri)?,
     })
 }
+
+fn build_schedule_execution_observation_request(
+    uri: &hyper::Uri,
+    principal: &AdminPrincipal,
+    scope: AdminFamilyScope,
+) -> Result<list::ScheduleExecutionObservationRequest, HttpError> {
+    let offset = parse_optional_u64_param(uri, "offset")?.unwrap_or_default();
+    Ok(list::ScheduleExecutionObservationRequest {
+        family: require_family_for_search(uri, principal, scope)?,
+        operation: list::parse_optional_string_query_param(uri, "operation"),
+        offset: usize::try_from(offset).map_err(|_| {
+            Box::new(error_response(
+                StatusCode::BAD_REQUEST,
+                "Query parameter 'offset' is too large",
+            ))
+        })?,
+        limit: parse_admin_record_limit(uri)?,
+    })
+}
+
+fn build_schedule_missed_observation_request(
+    uri: &hyper::Uri,
+    principal: &AdminPrincipal,
+    scope: AdminFamilyScope,
+) -> Result<list::ScheduleMissedObservationRequest, HttpError> {
+    Ok(list::ScheduleMissedObservationRequest {
+        family: require_family_for_search(uri, principal, scope)?,
+        realm: list::parse_optional_string_query_param(uri, "realm"),
+        area: list::parse_optional_string_query_param(uri, "area"),
+        resource: list::parse_optional_string_query_param(uri, "resource"),
+        operation: list::parse_optional_string_query_param(uri, "operation"),
+        limit: parse_admin_record_limit(uri)?,
+    })
+}

src/api/admin/list.rs

@@ -43,13 +43,14 @@ pub use dto_operations::{
     KvTransactionsList, LeaseSearchItem, LeaseSearchResponse, LeaseWaiterInfo,
     NoticeDeliveryObservation, NoticeDeliveryObservationList, NoticeRouteInfo, NoticeRoutesList,
     NoticeSubscription, NoticeSubscriptionsList, RpcCallObservation, RpcCallObservationList,
-    ScheduleExecutionObservation, ScheduleExecutionObservationList, ScheduleLatencyBuckets,
-    ScheduleMissedObservation, ScheduleMissedObservationList, SchedulePendingClaimInfo,
-    StreamAdminRecord, StreamInfo, StreamLagBuckets, StreamLatencyBuckets, StreamRecordsResponse,
-    StreamsList,
+    ScheduleExecutionObservation, ScheduleExecutionObservationList,
+    ScheduleExecutionObservationRequest, ScheduleLatencyBuckets, ScheduleMissedObservation,
+    ScheduleMissedObservationList, SchedulePendingClaimInfo, StreamAdminRecord, StreamInfo,
+    StreamLagBuckets, StreamLatencyBuckets, StreamRecordsResponse, StreamsList,
 };
 pub(crate) use dto_operations::{
-    LeaseSearchRequest, RpcCallObservationRequest, StreamSearchRequest,
+    LeaseSearchRequest, RpcCallObservationRequest, ScheduleMissedObservationRequest,
+    StreamSearchRequest,
 };
 pub use dto_queue_runtime::{
     LeaseInfo, LeasesList, QueueAreaCollection, QueueAreaDetail, QueueAreaEntry, QueueDeadLetter,

src/api/admin/list/admin_reads/schedule.rs

@@ -1,67 +1,103 @@
 use super::super::{
     matches_family, route_quad, troubleshooting, ResourcePath, Response, RouteFamily, Runtime,
-    ScheduleExecutionObservation, ScheduleExecutionObservationList, ScheduleMissedObservation,
-    ScheduleMissedObservationList,
+    ScheduleExecutionObservation, ScheduleExecutionObservationList,
+    ScheduleExecutionObservationRequest, ScheduleMissedObservation, ScheduleMissedObservationList,
+    ScheduleMissedObservationRequest,
 };
 use super::timestamp_ms_to_rfc3339;
+use std::collections::HashMap;
 
 /// Returns current schedule execution observations for the given resource.
 ///
 /// # Errors
 ///
 /// Propagates JSON response construction failures from the admin HTTP layer.
+///
+/// # Panics
+///
+/// Panics if the route family was not validated at the HTTP boundary.
 #[must_use]
 pub fn schedule_executions_for_resource(
     runtime: &Runtime,
     path: &ResourcePath<'_>,
-    family: u64,
-    limit: usize,
+    request: &ScheduleExecutionObservationRequest,
 ) -> Response {
-    let observations = runtime
+    let family = RouteFamily::try_from(request.family)
+        .expect("admin route family is validated at the HTTP boundary");
+    let pending_handoffs = runtime
+        .schedule_list_pending_claims(family)
+        .into_iter()
+        .filter_map(|claim| {
+            let route = route_quad(&claim.route)?;
+            (path.matches(route.realm, route.area, route.resource)
+                && request
+                    .operation
+                    .as_deref()
+                    .is_none_or(|expected| route.operation == expected))
+            .then(|| route.operation.to_string())
+        })
+        .fold(HashMap::<String, usize>::new(), |mut counts, operation| {
+            let count = counts.entry(operation).or_default();
+            *count = count.saturating_add(1);
+            counts
+        });
+    let mut observations = runtime
         .schedule_list_schedules(Some(path.realm))
         .into_iter()
         .filter(|schedule| {
-            schedule.route_family == family
+            schedule.route_family == request.family
                 && path.matches(&schedule.realm, &schedule.area, &schedule.resource)
+                && request
+                    .operation
+                    .as_deref()
+                    .is_none_or(|expected| schedule.operation == expected)
         })
-        .take(limit)
-        .map(|schedule| ScheduleExecutionObservation {
-            route_family: schedule.route_family,
-            realm: schedule.realm,
-            area: schedule.area,
-            resource: schedule.resource,
-            operation: schedule.operation,
-            delivery_mode: schedule.delivery_mode,
-            status: if schedule.last_run.is_some() {
-                "acknowledged_handoff".to_string()
-            } else {
-                "scheduled".to_string()
-            },
-            cron: schedule.cron,
-            next_run: schedule.next_run,
-            last_run: schedule.last_run,
-            executions_total: schedule.executions_total,
+        .skip(request.offset)
+        .take(request.limit.saturating_add(1))
+        .map(|schedule| {
+            let pending_handoffs = pending_handoffs
+                .get(&schedule.operation)
+                .copied()
+                .unwrap_or_default();
+            ScheduleExecutionObservation {
+                route_family: schedule.route_family,
+                realm: schedule.realm,
+                area: schedule.area,
+                resource: schedule.resource,
+                operation: schedule.operation,
+                delivery_mode: schedule.delivery_mode,
+                status: if schedule.last_run.is_some() {
+                    "acknowledged_handoff".to_string()
+                } else {
+                    "scheduled".to_string()
+                },
+                cron: schedule.cron,
+                next_run: schedule.next_run,
+                last_run: schedule.last_run,
+                executions_total: schedule.executions_total,
+                pending_handoffs,
+            }
         })
-        .collect();
+        .collect::<Vec<_>>();
+    let has_more = observations.len() > request.limit;
+    observations.truncate(request.limit);
 
     crate::api::admin::json_response(ScheduleExecutionObservationList {
-        route_family: family,
+        route_family: request.family,
         realm: path.realm.to_string(),
         area: path.area.to_string(),
         resource: path.resource.to_string(),
-        limit,
+        offset: request.offset,
+        limit: request.limit,
+        has_more,
         observations,
     })
 }
 
 /// Returns pending schedule handoff observations for the requested scope.
 pub(crate) fn schedule_missed_observations(
     runtime: &Runtime,
-    family: u64,
-    realm: Option<&str>,
-    area: Option<&str>,
-    resource: Option<&str>,
-    limit: usize,
+    request: &ScheduleMissedObservationRequest,
 ) -> Response {
     let now_ms = u64::try_from(
         std::time::SystemTime::now()
@@ -72,18 +108,31 @@ pub(crate) fn schedule_missed_observations(
     .unwrap_or(u64::MAX);
     let observations = runtime
         .schedule_list_pending_claims(
-            RouteFamily::try_from(family)
+            RouteFamily::try_from(request.family)
                 .expect("admin route family is validated at the HTTP boundary"),
         )
         .into_iter()
         .filter_map(|claim| {
             let route = route_quad(&claim.route)?;
-            if realm.is_none_or(|value| route.realm == value)
-                && area.is_none_or(|value| route.area == value)
-                && resource.is_none_or(|value| route.resource == value)
+            if request
+                .realm
+                .as_deref()
+                .is_none_or(|value| route.realm == value)
+                && request
+                    .area
+                    .as_deref()
+                    .is_none_or(|value| route.area == value)
+                && request
+                    .resource
+                    .as_deref()
+                    .is_none_or(|value| route.resource == value)
+                && request
+                    .operation
+                    .as_deref()
+                    .is_none_or(|value| route.operation == value)
             {
                 Some(ScheduleMissedObservation {
-                    route_family: family,
+                    route_family: request.family,
                     realm: route.realm.to_string(),
                     area: route.area.to_string(),
                     resource: route.resource.to_string(),
@@ -99,12 +148,12 @@ pub(crate) fn schedule_missed_observations(
                 None
             }
         })
-        .take(limit)
+        .take(request.limit)
         .collect();
 
     crate::api::admin::json_response(ScheduleMissedObservationList {
-        route_family: family,
-        limit,
+        route_family: request.family,
+        limit: request.limit,
         observations,
     })
 }