// Copyright 2024 The NativeLink Authors. All rights reserved.

//

// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

//    See LICENSE file for details

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

use core::sync::atomic::{AtomicUsize, Ordering};

use std::collections::{HashMap, HashSet, VecDeque};

use std::sync::Arc;

use std::time::{Instant, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;

use futures::{Future, FutureExt, StreamExt, future};

use nativelink_config::schedulers::SimpleSpec;

use nativelink_error::{Code, Error, ResultExt, make_err};

use nativelink_metric::{MetricsComponent, RootMetricsComponent};

use nativelink_proto::com::github::trace_machina::nativelink::events::{

    Event, OriginEvent, RequestEvent, event, request_event,

};

use nativelink_proto::com::github::trace_machina::nativelink::remote_execution::{

    StartExecute, WorkerLoad,

};

use nativelink_util::action_messages::{ActionInfo, ActionState, OperationId, WorkerId};

use nativelink_util::instant_wrapper::InstantWrapper;

use nativelink_util::metrics::{

    AgeBuckets, record_awaited_action_orphan, record_executing_ages, record_matching_pass,

    record_parked_dispatched, record_queue_ages, record_queue_depth, record_sweep_failure,

    record_unsatisfiable_failed, record_unsatisfiable_queued,

};

use nativelink_util::operation_state_manager::{

    ActionStateResult, ActionStateResultStream, ClientStateManager, MatchingEngineStateManager,

    OperationFilter, OperationStageFlags, OrderDirection, UpdateOperationType,

};

use nativelink_util::origin_event::{OriginMetadata, get_node_id};

use nativelink_util::platform_properties::PlatformProperties;

use nativelink_util::shutdown_guard::ShutdownGuard;

use nativelink_util::spawn;

use nativelink_util::task::JoinHandleDropGuard;

use opentelemetry::KeyValue;

use opentelemetry::baggage::BaggageExt;

use opentelemetry::context::{Context, FutureExt as OtelFutureExt};

use opentelemetry_semantic_conventions::attribute::ENDUSER_ID;

use parking_lot::Mutex;

use tokio::sync::{Notify, mpsc};

use tokio::time::Duration;

use tracing::{debug, error, info, info_span, warn};

use uuid::Uuid;

use crate::api_worker_scheduler::{ApiWorkerScheduler, QueuedAction};

use crate::awaited_action_db::{AwaitedActionDb, CLIENT_KEEPALIVE_DURATION, PeerScheduler};

use crate::known_platform_property_provider::KnownPlatformPropertyProvider;

use crate::match_outcome::{MatchOutcome, PropertyShape, UnsatisfiableReason};

use crate::platform_property_manager::PlatformPropertyManager;

use crate::simple_scheduler_state_manager::{

    RetryLimits, SimpleSchedulerStateManager, is_lost_record,

};

use crate::unsatisfiable_tracker::UnsatisfiableTracker;

use crate::worker::{ActionInfoWithProps, Worker, WorkerTimestamp};

use crate::worker_registry::WorkerRegistry;

use crate::worker_scheduler::{WorkerScheduler, WorkerSummary};

/// Default timeout for workers in seconds: four default keepalive intervals

/// (2.5s each). The old value of two meant one late keepalive under load

/// evicted the worker and requeued everything it held.

/// If this changes, remember to change the documentation in the config.

const DEFAULT_WORKER_TIMEOUT_S: u64 = 10;

/// How often this scheduler tells its peers what its workers can run and

/// reads what theirs can.

const FLEET_EXCHANGE_INTERVAL: Duration = Duration::from_secs(5);

/// How many exchange intervals a published fleet record outlives its last

/// refresh, so one missed refresh does not make a peer's workers vanish.

const FLEET_RECORD_TTL_INTERVALS: u32 = 3;

/// Floor on how many queued actions one matching pass fails as unsatisfiable.

/// The cap bleeds a large burst over several (back-to-back) passes instead of

/// failing a whole shape-class at once, and bounds the cumulative

/// version-conflict backoff sleeps a single pass can incur.

const MAX_UNSATISFIABLE_FAILS_PER_PASS: usize = 16;

/// A backlog larger than `floor * DIVISOR` raises the cap to `backlog /

/// DIVISOR`, so a large single-shape burst drains in ~DIVISOR back-to-back

/// passes rather than `backlog / floor` of them (whose tail could outlast a

/// client's own timeout).

const UNSATISFIABLE_FAIL_BACKLOG_DIVISOR: usize = 4;

/// The per-pass unsatisfiable-fail cap for a pass, given the previous pass's

/// unsatisfiable backlog: the floor, or a fraction of the backlog if larger.

const fn unsatisfiable_fail_cap(last_backlog: usize) -> usize {

    let scaled = last_backlog / UNSATISFIABLE_FAIL_BACKLOG_DIVISOR;

    if scaled > MAX_UNSATISFIABLE_FAILS_PER_PASS {

        scaled

    } else {

        MAX_UNSATISFIABLE_FAILS_PER_PASS

    }

}

/// Default timeout for a sent kill to be acknowledged in seconds.

/// If this changes, remember to change the documentation in the config.

const DEFAULT_UNACKNOWLEDGED_KILL_TIMEOUT_S: u64 = 60;

/// Mark operations as completed with error if no client has updated them

/// within this duration.

/// If this changes, remember to change the documentation in the config.

pub(crate) const DEFAULT_CLIENT_ACTION_TIMEOUT_S: u64 = 60;

/// Default times a job can retry before failing.

/// If this changes, remember to change the documentation in the config.

const DEFAULT_MAX_JOB_RETRIES: usize = 3;

struct SimpleSchedulerActionStateResult {

    client_operation_id: OperationId,

    action_state_result: Box<dyn ActionStateResult>,

}

impl SimpleSchedulerActionStateResult {

    fn new(

        client_operation_id: OperationId,

        action_state_result: Box<dyn ActionStateResult>,

    ) -> Self {

        Self {

            client_operation_id,

            action_state_result,

        }

    }

}

#[async_trait]

impl ActionStateResult for SimpleSchedulerActionStateResult {

    async fn as_state(&self) -> Result<(Arc<ActionState>, Option<OriginMetadata>), Error> {

        let (mut action_state, origin_metadata) = self

            .action_state_result

            .as_state()

            .await

            .err_tip(|| "In SimpleSchedulerActionStateResult")?;

        // We need to ensure the client is not aware of the downstream

        // operation id, so override it before it goes out.

        Arc::make_mut(&mut action_state).client_operation_id = self.client_operation_id.clone();

        Ok((action_state, origin_metadata))

    }

    async fn changed(&mut self) -> Result<(Arc<ActionState>, Option<OriginMetadata>), Error> {

        let (mut action_state, origin_metadata) = self

            .action_state_result

            .changed()

            .await

            .err_tip(|| "In SimpleSchedulerActionStateResult")?;

        // We need to ensure the client is not aware of the downstream

        // operation id, so override it before it goes out.

        Arc::make_mut(&mut action_state).client_operation_id = self.client_operation_id.clone();

        Ok((action_state, origin_metadata))

    }

    async fn as_action_info(&self) -> Result<(Arc<ActionInfo>, Option<OriginMetadata>), Error> {

        self.action_state_result

            .as_action_info()

            .await

            .err_tip(|| "In SimpleSchedulerActionStateResult")

    }

}

/// Engine used to manage the queued/running tasks and relationship with

/// the worker nodes. All state on how the workers and actions are interacting

/// should be held in this struct.

#[derive(MetricsComponent)]

pub struct SimpleScheduler {

    /// Manager for matching engine side of the state manager.

    #[metric(group = "matching_engine_state_manager")]

    matching_engine_state_manager: Arc<dyn MatchingEngineStateManager>,

    /// Manager for client state of this scheduler.

    #[metric(group = "client_state_manager")]

    client_state_manager: Arc<dyn ClientStateManager>,

    /// Manager for platform of this scheduler.

    #[metric(group = "platform_properties")]

    platform_property_manager: Arc<PlatformPropertyManager>,

    /// A `Workers` pool that contains all workers that are available to execute actions in a priority

    /// order based on the allocation strategy.

    #[metric(group = "worker_scheduler")]

    worker_scheduler: Arc<ApiWorkerScheduler>,

    /// The sender to send origin events to the origin events.

    maybe_origin_event_tx: Option<mpsc::Sender<OriginEvent>>,

    /// Background task that tries to match actions to workers. If this struct

    /// is dropped the spawn will be cancelled as well.

    task_worker_matching_spawn: JoinHandleDropGuard<()>,

    /// Background task that retires queued actions nobody is waiting on.

    /// Dropped with the struct, like the matching task.

    task_abandoned_sweep_spawn: JoinHandleDropGuard<()>,

    /// Background task that retires executing actions whose worker has gone

    /// quiet or no longer runs them. Dropped with the struct; `None` when

    /// `executing_sweep_interval_s` disables it.

    task_executing_sweep_spawn: Option<JoinHandleDropGuard<()>>,

    /// Background task that swaps worker capabilities with peer schedulers

    /// on the same state. Dropped with the struct, like the matching task.

    task_fleet_exchange_spawn: JoinHandleDropGuard<()>,

    /// Every duration, do logging of worker matching

    /// e.g. "worker busy", "can't find any worker"

    /// Set to None to disable. This is quite noisy, so we limit it

    worker_match_logging_interval: Option<Duration>,

    /// How long each action property shape has gone without any worker able

    /// to run it.

    unsatisfiable_tracker: Mutex<UnsatisfiableTracker>,

    /// The clock for actions queued while no worker is connected anywhere

    /// (`no_worker_action_timeout_s`); `None` when that timeout is off.

    no_worker_tracker: Option<Mutex<UnsatisfiableTracker>>,

    /// Typed platform properties by the hash of the raw ones. Typing is a

    /// lookup and a parse per property per queued action per pass, and the

    /// raw properties of a queued action rarely change (an escalation is

    /// the exception, and it changes the hash), so the pass reuses the

    /// typed set. Cleared when full rather than evicted, like

    /// `static_verdicts`.

    typed_properties: Mutex<HashMap<u64, PlatformProperties>>,

    /// The number of unsatisfiable actions still queued at the end of the last

    /// matching pass, used to size the next pass's per-pass fail cap so a large

    /// burst drains in a few passes. See `unsatisfiable_fail_cap`.

    last_unsatisfiable_backlog: AtomicUsize,

    /// The clock the state manager uses.

    now_fn: Box<dyn Fn() -> SystemTime + Send + Sync>,

}

impl core::fmt::Debug for SimpleScheduler {

    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {

        f.debug_struct("SimpleScheduler")

            .field("platform_property_manager", &self.platform_property_manager)

            .field("worker_scheduler", &self.worker_scheduler)

            .field("maybe_origin_event_tx", &self.maybe_origin_event_tx)

            .field(

                "task_worker_matching_spawn",

                &self.task_worker_matching_spawn,

            )

            .field(

                "task_abandoned_sweep_spawn",

                &self.task_abandoned_sweep_spawn,

            )

            .field(

                "task_executing_sweep_spawn",

                &self.task_executing_sweep_spawn,

            )

            .field("task_fleet_exchange_spawn", &self.task_fleet_exchange_spawn)

            .finish_non_exhaustive()

    }

}

/// How many typed property sets the pass keeps before starting over.

const MAX_TYPED_PROPERTIES: usize = 10_000;

/// The typed form of an action's raw properties, from the cache when the

/// same raw set was typed before.

fn typed_properties_for(

    cache: &Mutex<HashMap<u64, PlatformProperties>>,

    manager: &PlatformPropertyManager,

    raw: &HashMap<String, String>,

) -> Result<PlatformProperties, Error> {

    use core::hash::{Hash, Hasher};

    let mut pairs: Vec<(&String, &String)> = raw.iter().collect();

    pairs.sort();

    let mut hasher = std::collections::hash_map::DefaultHasher::new();

    for (name, value) in 
&pairs588
 {

        name.hash(&mut hasher);

        value.hash(&mut hasher);

    }

    let key = hasher.finish();

    if let Some(
typed443
) = cache.lock().get(&key) {

        return Ok(typed.clone());

    }

    let typed = manager.make_platform_properties(raw.clone())
?0
;

    let mut cache = cache.lock();

    if cache.len() >= MAX_TYPED_PROPERTIES {

        cache.clear();

    }

    cache.insert(key, typed.clone());

    Ok(typed)

}

impl SimpleScheduler {

    fn origin_event_id(event: &Event) -> String {

        Uuid::now_v6(&get_node_id(Some(event)))

            .hyphenated()

            .to_string()

    }

    fn scheduler_start_execute_event(

        worker_id: &WorkerId,

        operation_id: &OperationId,

        action_info: &ActionInfoWithProps,

    ) -> Event {

        let start_execute = StartExecute {

            execute_request: Some(action_info.inner.as_ref().into()),

            operation_id: operation_id.to_string(),

            queued_timestamp: Some(action_info.inner.insert_timestamp.into()),

            platform: Some((&action_info.platform_properties).into()),

            worker_id: worker_id.to_string(),

            request_metadata: action_info.origin_metadata.bazel_metadata.clone(),

        };

        Event {

            event: Some(event::Event::Request(RequestEvent {

                event: Some(request_event::Event::SchedulerStartExecute(start_execute)),

            })),

        }

    }

    async fn publish_scheduler_start_execute(

        maybe_origin_event_tx: Option<&mpsc::Sender<OriginEvent>>,

        origin_metadata: &OriginMetadata,

        event_id: String,

        event: Event,

    ) {

        let Some(origin_event_tx) = maybe_origin_event_tx else {

            return;

        };

        let origin_event = OriginEvent {

            version: 0,

            event_id,

            parent_event_id: String::new(),

            bazel_request_metadata: origin_metadata.bazel_metadata.clone(),

            identity: origin_metadata.identity.clone(),

            event: Some(event),

        };

        // Awaited send (not try_send): backpressure rather than drop, so the

        // start-execute event that later resource-usage events reference as

        // their parent isn't silently lost when the queue is full.

        if let Err(
err0
) = origin_event_tx.send(origin_event).await {

            warn!(

                ?err,

                "Failed to publish scheduler start execute origin event"

            );

        }

    }

    /// Attempts to find a worker to execute an action and begins executing it.

    /// If an action is already running that is cacheable it may merge this

    /// action with the results and state changes of the already running

    /// action. If the task cannot be executed immediately it will be queued

    /// for execution based on priority and other metrics.

    /// All further updates to the action will be provided through the returned

    /// value.

    async fn inner_add_action(

        &self,

        client_operation_id: OperationId,

        action_info: Arc<ActionInfo>,

    ) -> Result<Box<dyn ActionStateResult>, Error> {

        let action_state_result = self

            .client_state_manager

            .add_action(client_operation_id.clone(), action_info)

            .await

            .err_tip(|| "In SimpleScheduler::add_action")
?0
;

        Ok(Box::new(SimpleSchedulerActionStateResult::new(

            client_operation_id.clone(),

            action_state_result,

        )))

    }

    async fn inner_filter_operations(

        &self,

        filter: OperationFilter,

    ) -> Result<ActionStateResultStream<'_>, Error> {

        self.client_state_manager

            .filter_operations(filter)

            .await

            .err_tip(|| "In SimpleScheduler::find_by_client_operation_id getting filter result")

    }

    async fn get_queued_operations(&self) -> Result<ActionStateResultStream<'_>, Error> {

        let filter = OperationFilter {

            stages: OperationStageFlags::Queued,

            order_by_priority_direction: Some(OrderDirection::Desc),

            ..Default::default()

        };

        self.matching_engine_state_manager

            .filter_operations(filter)

            .await

            .err_tip(|| "In SimpleScheduler::get_queued_operations getting filter result")

    }

    pub async fn do_try_match_for_test(&self) -> Result<(), Error> {

        self.do_try_match(true).await

    }

    /// A pass as production runs it between logging intervals: cached

    /// verdicts are used rather than recomputed.

    pub async fn do_try_match_quietly_for_test(&self) -> Result<(), Error> {

        self.do_try_match(false).await

    }

    /// Stands in for the fleet exchange with peer schedulers: one peer with

    /// these workers.

    
pub async fn set_peer_fleet_for_test(&self, peer_fleet: Vec<PlatformProperties>)0
 {

        self.set_peer_schedulers_for_test(vec![PeerScheduler {

            scheduler_id: "peer".to_string(),

            workers: peer_fleet,

        }])

        .await;

    }

    /// Stands in for the fleet exchange with peer schedulers.

    pub async fn set_peer_schedulers_for_test(&self, peers: Vec<PeerScheduler>) {

        self.worker_scheduler

            .set_peer_fleet(Some(peers), (self.now_fn)())

            .await;

    }

    /// The id this scheduler publishes to its peers under.

    pub fn scheduler_id_for_test(&self) -> &str {

        self.worker_scheduler.scheduler_id()

    }

    /// The workers held for a queued action right now.

    
pub async fn held_workers_for_test(&self) -> Vec<WorkerId>0
 {

        self.worker_scheduler.held_workers_for_test().await

    }

    /// Stands in for a failed fleet exchange with peer schedulers.

    
pub async fn set_peers_unknown_for_test(&self)0
 {

        self.worker_scheduler

            .set_peer_fleet(None, (self.now_fn)())

            .await;

    }

    /// A counter that changes every time the fleet does.

    
pub async fn fleet_generation_for_test(&self) -> u640
 {

        self.worker_scheduler.fleet_generation().await

    }

    /// What this scheduler would publish to peer schedulers.

    
pub async fn fleet_shapes_for_test(&self) -> Vec<PlatformProperties>0
 {

        self.worker_scheduler.fleet_shapes().await

    }

    /// How long until the matching loop would next wake on behalf of an

    /// unsatisfiable shape, if at all.

    pub fn unsatisfiable_deadline_for_test(&self) -> Option<Duration> {

        self.unsatisfiable_deadline((self.now_fn)())

    }

    /// How long until a pass should run for an unsatisfiable or a no-worker

    /// shape, whichever comes first.

    fn unsatisfiable_deadline(&self, now: SystemTime) -> Option<Duration> {

        let unsatisfiable = self.unsatisfiable_tracker.lock().next_deadline(now);

        let no_worker = self

            .no_worker_tracker

            .as_ref()

            .and_then(|tracker| 
tracker.lock()6
.
next_deadline6
(
now6
));

        match (unsatisfiable, no_worker) {

            (Some(a), Some(b)) => Some(a.min(b)),

            (a, b) => a.or(b),

        }

    }

    // TODO(palfrey) This is an O(n*m) (aka n^2) algorithm. In theory we

    // can create a map of capabilities of each worker and then try and match

    // the actions to the worker using the map lookup (ie. map reduce).

    async fn do_try_match(&self, full_worker_logging: bool) -> Result<(), Error> {

        let match_started = Instant::now();

        let 
result859
 = self.do_try_match_inner(full_worker_logging).await;

        record_matching_pass(match_started.elapsed().as_secs_f64(), result.is_ok());

        result

    }

    async fn do_try_match_inner(&self, full_worker_logging: bool) -> Result<(), Error> {

        struct UnsatisfiablePass<'a> {

            tracker: &'a Mutex<UnsatisfiableTracker>,

            now: SystemTime,

            fleet_generation: u64,

            /// Most actions this pass may fail as unsatisfiable (blast-radius

            /// cap, sized from the last pass's backlog by `unsatisfiable_fail_cap`).

            cap: usize,

            /// How many it has failed so far this pass.

            fails_this_pass: AtomicUsize,

        }

        /// Logs an action no worker can run and, once its property shape has

        /// been unsatisfiable for the configured timeout and the action has

        /// itself been queued that long, fails it.

        async fn handle_unsatisfiable(

            action_state_result: &dyn ActionStateResult,

            matching_engine_state_manager: &dyn MatchingEngineStateManager,

            workers: &ApiWorkerScheduler,

            platform_properties: &PlatformProperties,

            insert_timestamp: SystemTime,

            reason: &UnsatisfiableReason,

            pass: &UnsatisfiablePass<'_>,

        ) -> Result<(), Error> {

            let shape = PropertyShape::from(platform_properties);

            let (observation, fails_actions) = {

                let mut tracker = pass.tracker.lock();

                let observation = tracker.observe(

                    shape.clone(),

                    pass.now,

                    pass.fleet_generation,

                    insert_timestamp,

                );

                (observation, tracker.fails_actions())

            };

            if observation.should_warn {

                // Without a timeout nothing is failed, and a pool that

                // scales up from zero shows up here on every cold start.

                if fails_actions {

                    warn!(

                        %reason,

                        waited_s = observation.waited.as_secs(),

                        "Queued action cannot run on any connected worker"

                    );

                } else {

                    info!(

                        %reason,

                        waited_s = observation.waited.as_secs(),

                        "Queued action cannot run on any connected worker"

                    );

                }

            }

            if !observation.is_due {

                return Ok(());

            }

            // The action must also have waited the timeout itself, so one that

            // arrives while its shape is already due is not failed on sight.

            // Actions queued together still fail within about one timeout;

            // during a pool outage later actions fail as they age instead of

            // in an immediate burst. The tracker wakes the loop when it does.

            if !observation.action_eligible {

                return Ok(());

            }

            // A worker that joined here or on a peer since the pass began may

            // be able to run it, so leave it for the next pass to judge.

            if workers.fleet_generation().await != pass.fleet_generation {

                return Ok(());

            }

            // Blast-radius cap: fail at most `cap` actions per pass. Over-cap

            // actions were still observed above (their timers keep running) and

            // are reconsidered next pass, so a large burst bleeds off over a few

            // back-to-back passes rather than failing all at once.

            if pass.fails_this_pass.load(Ordering::Relaxed) >= pass.cap {

                return Ok(());

            }

            pass.fails_this_pass.fetch_add(1, Ordering::Relaxed);

            let operation_id = {

                let (action_state, _origin_metadata) = action_state_result

                    .as_state()

                    .await

                    .err_tip(|| "Failed to get state of an unsatisfiable action")
?0
;

                action_state.client_operation_id.clone()

            };

            // The client only hears what it asked for; what workers offer

            // stays in the server log above.

            let err = make_err!(

                Code::FailedPrecondition,

                "Action cannot be scheduled, {}. No capable worker has been connected for {}s.",

                reason.for_client(),

                observation.waited.as_secs()

            );

            let failed = matching_engine_state_manager

                .fail_queued_operation(&operation_id, err)

                .await

                .err_tip(|| "Failed to fail an unsatisfiable action in do_try_match")
?0
;

            if failed {

                record_unsatisfiable_failed(reason.property_names());

                // So a pass that retires every eligible action of the shape

                // does not read as stuck and back off the next wake.

                pass.tracker.lock().note_failed(&shape);

            
}0

            Ok(())

        }

        /// A listed action with what matching needs of it.

        struct Loaded {

            action_state_result: Box<dyn ActionStateResult>,

            action_info: ActionInfoWithProps,

        }

        /// Reads what matching needs of a listed action. `None` when the

        /// record cannot be read: gone from the store (eviction), or

        /// unreadable. Counted so an operator can see the store losing

        /// records, skipped so the rest of the pass still runs, and not an

        /// error, since an error here made the pass rerun at once and log

        /// ten times a second until the entry went away.

        async fn load_action(

            action_state_result: Box<dyn ActionStateResult>,

            platform_property_manager: &PlatformPropertyManager,

            typed_properties: &Mutex<HashMap<u64, PlatformProperties>>,

        ) -> Result<Option<Loaded>, Error> {

            let (action_info, maybe_origin_metadata) =

                match action_state_result.as_action_info().await {

                    Ok(found) => found,

                    Err(
err0
) if is_lost_record(&err
)0
 => {

                        if err.code == Code::NotFound {

                            debug!(

                                ?err,

                                "Queued operation listed but its record is gone; skipping"

                            );

                        } else {

                            warn!(

                                ?err,

                                "Queued operation listed but its record cannot be read; skipping"

                            );

                        }

                        record_awaited_action_orphan("matching");

                        return Ok(None);

                    }

                    // The store itself failed, so the pass is incomplete:

                    // an error here reruns it at once, as before.

                    Err(err) => {

                        return Err(err).err_tip(

                            || "Failed to get action info in SimpleScheduler::do_try_match",

                        );

                    }

                };

            let platform_properties = typed_properties_for(

                typed_properties,

                platform_property_manager,

                &action_info.platform_properties,

            )

            .err_tip(|| "Failed to make platform properties in SimpleScheduler::do_try_match")
?0
;

            Ok(Some(Loaded {

                action_state_result,

                action_info: ActionInfoWithProps {

                    inner: action_info,

                    platform_properties,

                    origin_metadata: maybe_origin_metadata.unwrap_or_default(),

                    scheduler_start_execute_event_id: None,

                },

            }))

        }

        /// What became of a dispatch that did not fail.

        #[derive(Clone, Copy, PartialEq, Eq)]

        enum Dispatch {

            /// The worker was told to run the action.

            Sent,

            /// Another scheduler assigned the action first; nothing was sent.

            LostRace,

        }

        /// Hands a matched action to its worker. `Ok(Dispatch::Sent)` means

        /// the worker was actually told; a lost assignment race is not an

        /// error but sends nothing.

        async fn dispatch_to_worker(

            action_state_result: &dyn ActionStateResult,

            workers: &ApiWorkerScheduler,

            matching_engine_state_manager: &dyn MatchingEngineStateManager,

            maybe_origin_event_tx: Option<&mpsc::Sender<OriginEvent>>,

            worker_id: WorkerId,

            action_info: ActionInfoWithProps,

            now_fn: &(dyn Fn() -> SystemTime + Send + Sync),

        ) -> Result<Dispatch, Error> {

            let origin_metadata = action_info.origin_metadata.clone();

            let event_origin_metadata = origin_metadata.clone();

            let attach_operation_fut = async move {

                // Extract the operation_id from the action_state.

                let operation_id = {

                    let (action_state, _origin_metadata) = action_state_result

                        .as_state()

                        .await

                        .err_tip(|| "Failed to get action_info from as_state_result stream")
?0
;

                    action_state.client_operation_id.clone()

                };

                // Tell the matching engine that the operation is being assigned to a worker.

                let assign_result = matching_engine_state_manager

                    .assign_operation(&operation_id, Ok(&worker_id))

                    .await

                    .err_tip(|| "Failed to assign operation in do_try_match");

                if let Err(
err10
) = assign_result {

                    if err.code == Code::Aborted {

                        // If the operation was aborted, it means that the operation was

                        // cancelled due to another operation being assigned to the worker.

                        return Ok(Dispatch::LostRace);

                    }

                    // Any other error is a real error.

                    return Err(err);

                }

                let mut action_info = action_info;

                let scheduler_start_execute_event = maybe_origin_event_tx.map(|_| 
{1

                    let event = SimpleScheduler::scheduler_start_execute_event(

                        &worker_id,

                        &operation_id,

                        &action_info,

                    );

                    let event_id = SimpleScheduler::origin_event_id(&event);

                    action_info.scheduler_start_execute_event_id = Some(event_id.clone());

                    (event_id, event)

                });

                debug!(%worker_id, %operation_id, ?action_info, "Notifying worker of operation");

                let dispatched_at = now_fn()

                    .duration_since(UNIX_EPOCH)

                    .unwrap_or_default()

                    .as_secs();

                workers

                    .worker_notify_run_action(worker_id, operation_id, action_info, dispatched_at)

                    .await

                    .err_tip(|| 
{1

                        "Failed to run worker_notify_run_action in SimpleScheduler::do_try_match"

                    })?;

                if let Some((
event_id1
, 
event1
)) = scheduler_start_execute_event {

                    SimpleScheduler::publish_scheduler_start_execute(

                        maybe_origin_event_tx,

                        &event_origin_metadata,

                        event_id,

                        event,

                    )

                    .await;

                }

                Ok(Dispatch::Sent)

            };

            tokio::pin!(attach_operation_fut);

            let ctx = Context::current_with_baggage(vec![KeyValue::new(

                ENDUSER_ID,

                origin_metadata.identity,

            )]);

            info_span!("do_try_match")

                .in_scope(|| attach_operation_fut)

                .with_context(ctx)

                .await

        }

        /// Actions this pass read that found no room, in the order the

        /// queue listed them: highest priority first, then oldest. That is

        /// the order room is offered to them in, before the action the

        /// pass is reading, whenever the fleet's capacity changes. Parking

        /// is a push; a replay walks the list once and asks the fleet once

        /// per property shape it meets, since every action of one shape

        /// asks the same of a worker: a shape whose first parked action

        /// finds no room is skipped for the rest of that replay.

        #[derive(Default)]

        struct Parked {

            /// Each with its shape, computed once at parking.

            queue: VecDeque<(PropertyShape, Loaded)>,

            /// Actions that were not parked because the list was full.

            cap_dropped: u64,

        }

        /// Past this many parked actions a pass stops parking, so a newer

        /// action past that point may take room ahead of one of them. The

        /// head of the queue, which is what a client waits on, is always

        /// parked.

        const PARKED_CAP: usize = 10_000;

        impl Parked {

            fn park(&mut self, loaded: Loaded) {

                if self.queue.len() >= PARKED_CAP {

                    self.cap_dropped += 1;

                    return;

                }

                let shape = PropertyShape::from(&loaded.action_info.platform_properties);

                self.queue.push_back((shape, loaded));

            }

            fn is_empty(&self) -> bool {

                self.queue.is_empty()

            }

        }

        /// What one pass did with its parked actions.

        #[derive(Default)]

        struct Placement {

            /// Parked actions that found a worker and were dispatched to it.

            attempted: u64,

            /// Of those, the ones the worker was actually told to run.

            dispatched: u64,

            /// Of those, the ones another scheduler assigned first.

            lost_race: u64,

            /// Of those, the ones whose dispatch failed.

            failed: u64,

        }

        /// Offers the room that opened since the pass last looked to the

        /// parked actions in listing order. Every parked action that found

        /// a worker leaves the list, whatever its dispatch came to.

        #[expect(clippy::too_many_arguments)]

        async fn place_parked(

            parked: &mut Parked,

            placement: &mut Placement,

            contended: &mut HashSet<PropertyShape>,

            workers: &ApiWorkerScheduler,

            matching_engine_state_manager: &dyn MatchingEngineStateManager,

            maybe_origin_event_tx: Option<&mpsc::Sender<OriginEvent>>,

            full_worker_logging: bool,

            now: SystemTime,

            now_fn: &(dyn Fn() -> SystemTime + Send + Sync),

        ) -> Result<(), Error> {

            let mut result = Ok(());

            let mut blocked: HashSet<PropertyShape> = HashSet::new();

            let mut still_parked = VecDeque::with_capacity(parked.queue.len());

            while let Some((
shape7
, 
loaded7
)) = parked.queue.pop_front() {

                if blocked.contains(&shape) || contended.contains(&shape) {

                    still_parked.push_back((shape, loaded));

                    continue;

                }

                let MatchOutcome::Matched(
worker_id5
) = workers

                    .find_worker_for_action(

                        &loaded.action_info.platform_properties,

                        full_worker_logging,

                        now,

                        Some(QueuedAction {

                            unique_qualifier: &loaded.action_info.inner.unique_qualifier,

                            insert_timestamp: loaded.action_info.inner.insert_timestamp,

                        }),

                    )

                    .await

                else {

                    still_parked.push_back((shape.clone(), loaded));

                    blocked.insert(shape);

                    continue;

                };

                placement.attempted += 1;

                match dispatch_to_worker(

                    loaded.action_state_result.as_ref(),

                    workers,

                    matching_engine_state_manager,

                    maybe_origin_event_tx,

                    worker_id,

                    loaded.action_info,

                    now_fn,

                )

                .await

                {

                    Ok(Dispatch::Sent) => placement.dispatched += 1,

                    Ok(Dispatch::LostRace) => {

                        placement.lost_race += 1;

                        contended.insert(shape);

                    }

                    Err(err) => {

                        placement.failed += 1;

                        result = result.merge(Err(err));

                    }

                }

            }

            parked.queue = still_parked;

            result

        }

        let mut result = Ok(());

        let start = Instant::now();

        let unsatisfiable_pass = UnsatisfiablePass {

            tracker: &self.unsatisfiable_tracker,

            now: (self.now_fn)(),

            fleet_generation: self.worker_scheduler.fleet_generation().await,

            cap: unsatisfiable_fail_cap(self.last_unsatisfiable_backlog.load(Ordering::Relaxed)),

            fails_this_pass: AtomicUsize::new(0),

        };

        // With no worker connected anywhere and the no-worker timeout set,

        // every queued action is judged by that clock instead. A fleet the

        // census cannot vouch for counts as having workers.

        let no_worker_pass = match 
&self.no_worker_tracker10
 {

            Some(tracker)

                if self

                    .worker_scheduler

                    .no_worker_anywhere(unsatisfiable_pass.now)

                    .await =>

            {

                Some(UnsatisfiablePass {

                    tracker,

                    now: unsatisfiable_pass.now,

                    fleet_generation: unsatisfiable_pass.fleet_generation,

                    cap: unsatisfiable_pass.cap,

                    fails_this_pass: AtomicUsize::new(0),

                })

            }

            _ => None,

        };

        let no_workers_reason = UnsatisfiableReason::no_workers();

        let 
mut stream861
 = self

            .get_queued_operations()

            .await

            .err_tip(|| "Failed to get queued operations in do_try_match")
?0
;

        let query_elapsed = start.elapsed();

        if query_elapsed > Duration::from_secs(1) {

            warn!(

                elapsed_ms = query_elapsed.as_millis(),

                "Slow get_queued_operations query"

            );

        }

        let workers = self.worker_scheduler.as_ref();

        let matching_engine_state_manager = self.matching_engine_state_manager.as_ref();

        let maybe_origin_event_tx = self.maybe_origin_event_tx.as_ref();

        let mut parked = Parked::default();

        let mut placement = Placement::default();

        // Once a peer wins an assignment, let it advance through that

        // property shape for this pass. Other shapes can still use this

        // scheduler's workers, and the next pass tries every shape again.

        let mut contended = HashSet::new();

        let mut oldest_queued = Vec::new();

        let hold_pass = workers.begin_pass_holds().await;

        let mut capacity_seen = workers.capacity_generation();

        let mut queued_seen: u64 = 0;

        // How long each action read has waited since its client submitted

        // it, requeues included: the pass already reads every queued action,

        // so the ages cost nothing more from the store.

        let mut queued_ages = AgeBuckets::default();

        'actions: while let Some(
action_state_result589
) = stream.next().await {

            queued_seen += 1;

            let 
loaded588
 = match load_action(

                action_state_result,

                self.platform_property_manager.as_ref(),

                &self.typed_properties,

            )

            .await

            {

                Ok(Some(loaded)) => loaded,

                Ok(None) => continue,

                Err(err) => {

                    result = result.merge(Err(err));

                    continue;

                }

            };

            queued_ages.observe_since(

                loaded.action_info.inner.insert_timestamp,

                unsatisfiable_pass.now,

            );

            if full_worker_logging {

                oldest_queued.push(loaded.action_info.inner.clone());

                oldest_queued.sort_unstable_by_key(|action| 
action102
.insert_timestamp);

                oldest_queued.truncate(5);

            }

            let shape = PropertyShape::from(&loaded.action_info.platform_properties);

            if contended.contains(&shape) {

                continue;

            }

            // Room that opened since the pass last placed anything belongs

            // to the oldest action still waiting, not to this one, which is

            // newer than everything parked. The generation is read under the

            // placement lock, so a change cannot slip between the two.

            let outcome = loop {

                let (outcome, generation) = workers

                    .find_worker_for_action_observed(

                        &loaded.action_info.platform_properties,

                        full_worker_logging,

                        unsatisfiable_pass.now,

                        Some(QueuedAction {

                            unique_qualifier: &loaded.action_info.inner.unique_qualifier,

                            insert_timestamp: loaded.action_info.inner.insert_timestamp,

                        }),

                    )

                    .await;

                if generation == capacity_seen || 
parked5
.
is_empty5
() {

                    capacity_seen = generation;

                    break outcome;

                }

                capacity_seen = generation;

                result = result.merge(

                    place_parked(

                        &mut parked,

                        &mut placement,

                        &mut contended,

                        workers,

                        matching_engine_state_manager,

                        maybe_origin_event_tx,

                        full_worker_logging,

                        unsatisfiable_pass.now,

                        self.now_fn.as_ref(),

                    )

                    .await,

                );

                if contended.contains(&shape) {

                    continue 'actions;

                }

            };

            match outcome {

                MatchOutcome::Matched(worker_id) => {

                    result = result.merge(

                        dispatch_to_worker(

                            loaded.action_state_result.as_ref(),

                            workers,

                            matching_engine_state_manager,

                            maybe_origin_event_tx,

                            worker_id,

                            loaded.action_info,

                            self.now_fn.as_ref(),

                        )

                        .await

                        .map(|dispatch| 
{221

                            if dispatch == Dispatch::LostRace {

                                contended.insert(shape);

                            }

                        }),

                    );

                }

                // A worker could run it once one has room. The match

                // considered holding a worker for it (`reservation_hold`).

                MatchOutcome::WaitingForCapacity => {

                    parked.park(loaded);

                }

                // Nothing to wait for; a worker connecting starts a new pass,

                // unless the no-worker clock is running.

                MatchOutcome::NoWorkersConnected => {

                    if let Some(
no_worker_pass7
) = &no_worker_pass {

                        result = result.merge(

                            handle_unsatisfiable(

                                loaded.action_state_result.as_ref(),

                                matching_engine_state_manager,

                                workers,

                                &loaded.action_info.platform_properties,

                                loaded.action_info.inner.insert_timestamp,

                                &no_workers_reason,

                                no_worker_pass,

                            )

                            .await,

                        );