ironflow_api/
schedule_sync.rs

1//! Reconciliation of handler-declared schedules with the database.
2//!
3//! Call [`sync_handler_schedules`] then [`repair_unscheduled_schedules`] once
4//! at server startup, before spawning the
5//! [`ScheduleTicker`](super::schedule_ticker::ScheduleTicker).
6
7use std::collections::{HashMap, HashSet};
8
9use chrono::Utc;
10use ironflow_engine::engine::Engine;
11use ironflow_store::entities::{
12    MAX_PRIORITY, MIN_PRIORITY, NewSchedule, Schedule, ScheduleNext, ScheduleSource, ScheduleUpdate,
13};
14use ironflow_store::error::StoreError;
15use ironflow_store::store::Store;
16use serde_json::json;
17use tracing::{info, warn};
18
19use crate::schedule_ticker::schedule_next;
20
21/// Reconcile DB schedules with handler-declared schedules.
22///
23/// Performs a three-way sync at startup:
24/// 1. **Create** DB rows for handlers that declare a schedule but have no
25///    corresponding `source = handler` row.
26/// 2. **Update** the cron expression when the handler's cron changed, and the
27///    priority when the handler's
28///    [`priority`](ironflow_engine::handler::WorkflowHandler::priority) changed.
29/// 3. **Delete** orphan `source = handler` rows whose handler was removed
30///    from the code.
31///
32/// # Errors
33///
34/// Returns the first store error encountered. Changes applied before the
35/// error are kept.
36///
37/// # Examples
38///
39/// ```no_run
40/// use std::sync::Arc;
41/// use ironflow_api::schedule_sync::sync_handler_schedules;
42/// use ironflow_engine::engine::Engine;
43/// use ironflow_core::providers::claude::ClaudeCodeProvider;
44/// use ironflow_store::memory::InMemoryStore;
45/// use ironflow_store::store::Store;
46///
47/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
48/// let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
49/// let engine = Engine::new(store.clone(), Arc::new(ClaudeCodeProvider::new()));
50/// sync_handler_schedules(&engine, store.as_ref()).await?;
51/// # Ok(())
52/// # }
53/// ```
54pub async fn sync_handler_schedules(engine: &Engine, store: &dyn Store) -> Result<(), StoreError> {
55    let handlers = engine.scheduled_handlers();
56    // The priority every run of the schedule gets, clamped like a run the
57    // handler creates itself.
58    let handler_map: HashMap<&str, (&str, i16)> = handlers
59        .iter()
60        .map(|(name, cron)| {
61            let priority = engine.get_handler(name).map_or(0, |handler| {
62                handler.priority().clamp(MIN_PRIORITY, MAX_PRIORITY)
63            });
64            (*name, (cron.as_str(), priority))
65        })
66        .collect();
67    let handler_names: HashSet<&str> = handler_map.keys().copied().collect();
68
69    let existing = store.list_schedules(1, 1000).await?;
70
71    let handler_schedules: Vec<_> = existing
72        .items
73        .iter()
74        .filter(|s| s.source == ScheduleSource::Handler)
75        .collect();
76
77    // 1. Delete orphans: handler-declared schedules whose handler no longer exists.
78    for schedule in &handler_schedules {
79        if !handler_names.contains(schedule.workflow_name.as_str()) {
80            store.delete_schedule(schedule.id).await?;
81            info!(
82                workflow = %schedule.workflow_name,
83                "removed orphan handler schedule (handler no longer exists)"
84            );
85        }
86    }
87
88    // Build lookup of remaining handler schedules by workflow name.
89    let existing_map: HashMap<&str, &Schedule> = handler_schedules
90        .iter()
91        .filter(|s| handler_names.contains(s.workflow_name.as_str()))
92        .map(|s| (s.workflow_name.as_str(), *s))
93        .collect();
94
95    for (name, (cron_str, priority)) in &handler_map {
96        match existing_map.get(name) {
97            Some(existing) => {
98                let cron_changed = existing.cron_expression != *cron_str;
99                let priority_changed = existing.priority != *priority;
100                if !cron_changed && !priority_changed {
101                    continue;
102                }
103                let mut update = ScheduleUpdate::default();
104                if cron_changed {
105                    // 2. Cron changed in code: update DB row. A schedule Ironflow
106                    // disabled on an error is re-enabled by a cron that works.
107                    let next = schedule_next(cron_str);
108                    let reenable =
109                        existing.last_error.is_some() && matches!(next, ScheduleNext::At(_));
110                    update = next_trigger_update(next);
111                    update.cron_expression = Some(cron_str.to_string());
112                    if reenable {
113                        update.disabled_at = Some(None);
114                        update.last_error = Some(None);
115                    }
116                }
117                if priority_changed {
118                    update.priority = Some(*priority);
119                }
120                store.update_schedule(existing.id, update).await?;
121                info!(
122                    workflow = %name,
123                    old_cron = %existing.cron_expression,
124                    new_cron = %cron_str,
125                    old_priority = existing.priority,
126                    new_priority = *priority,
127                    "updated handler schedule"
128                );
129            }
130            None => {
131                // 3. Missing: create a new handler schedule.
132                let next = schedule_next(cron_str);
133                let next_trigger_at = match next {
134                    ScheduleNext::At(at) => Some(at),
135                    ScheduleNext::Disable { .. } => None,
136                };
137                let created = store
138                    .create_schedule(NewSchedule {
139                        workflow_name: name.to_string(),
140                        cron_expression: cron_str.to_string(),
141                        inputs: json!({}),
142                        source: ScheduleSource::Handler,
143                        priority: *priority,
144                        created_by_user_id: None,
145                        next_trigger_at,
146                    })
147                    .await?;
148                if let ScheduleNext::Disable { error } = next {
149                    warn!(
150                        workflow = %name,
151                        schedule = %cron_str,
152                        error = %error,
153                        "cannot compute next trigger, handler schedule disabled"
154                    );
155                    store
156                        .update_schedule(
157                            created.id,
158                            next_trigger_update(ScheduleNext::Disable { error }),
159                        )
160                        .await?;
161                }
162                info!(
163                    workflow = %name,
164                    schedule = %cron_str,
165                    "synced handler-declared schedule to DB"
166                );
167            }
168        }
169    }
170
171    Ok(())
172}
173
174/// The update that applies `next` to a schedule: its next trigger time, or
175/// disabled with the reason in `last_error`.
176fn next_trigger_update(next: ScheduleNext) -> ScheduleUpdate {
177    match next {
178        ScheduleNext::At(at) => ScheduleUpdate {
179            next_trigger_at: Some(Some(at)),
180            ..Default::default()
181        },
182        ScheduleNext::Disable { error } => ScheduleUpdate {
183            disabled_at: Some(Some(Utc::now())),
184            next_trigger_at: Some(None),
185            last_error: Some(Some(error)),
186            ..Default::default()
187        },
188    }
189}
190
191/// How many schedules [`repair_unscheduled_schedules`] reads per page.
192const REPAIR_PAGE_SIZE: u32 = 100;
193
194/// Repair active schedules that have no next trigger time.
195///
196/// Such a schedule never fires again. Earlier releases left schedules in that
197/// state when a run creation failed, when the server stopped between the
198/// claim and the update, or when the next trigger could not be computed.
199/// Each one receives its next trigger time, or is disabled with the reason in
200/// `last_error` when that time cannot be computed. Logs a `warn` per repaired
201/// schedule and returns how many were repaired.
202///
203/// Call it once at server startup, after
204/// [`sync_handler_schedules`] and before spawning the
205/// [`ScheduleTicker`](super::schedule_ticker::ScheduleTicker).
206///
207/// # Errors
208///
209/// Returns the first store error encountered. Repairs applied before the
210/// error are kept.
211///
212/// # Examples
213///
214/// ```no_run
215/// use ironflow_api::schedule_sync::repair_unscheduled_schedules;
216/// use ironflow_store::memory::InMemoryStore;
217///
218/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
219/// let store = InMemoryStore::new();
220/// let repaired = repair_unscheduled_schedules(&store).await?;
221/// println!("{repaired} schedules repaired");
222/// # Ok(())
223/// # }
224/// ```
225pub async fn repair_unscheduled_schedules(store: &dyn Store) -> Result<usize, StoreError> {
226    let mut broken: Vec<Schedule> = Vec::new();
227    let mut page = 1;
228    loop {
229        let batch = store.list_schedules(page, REPAIR_PAGE_SIZE).await?;
230        let read = (page as u64) * (REPAIR_PAGE_SIZE as u64);
231        let last = batch.items.is_empty() || read >= batch.total;
232        broken.extend(
233            batch
234                .items
235                .into_iter()
236                .filter(|s| s.is_active() && s.next_trigger_at.is_none()),
237        );
238        if last {
239            break;
240        }
241        page += 1;
242    }
243
244    for schedule in &broken {
245        let next = schedule_next(&schedule.cron_expression);
246        match &next {
247            ScheduleNext::At(at) => warn!(
248                schedule_id = %schedule.id,
249                workflow = %schedule.workflow_name,
250                next_trigger_at = %at,
251                "repaired active schedule without next trigger"
252            ),
253            ScheduleNext::Disable { error } => warn!(
254                schedule_id = %schedule.id,
255                workflow = %schedule.workflow_name,
256                error = %error,
257                "disabled active schedule without next trigger: cannot compute it"
258            ),
259        }
260        store
261            .update_schedule(schedule.id, next_trigger_update(next))
262            .await?;
263    }
264
265    Ok(broken.len())
266}
267
268#[cfg(test)]
269mod tests {
270    use ironflow_core::providers::claude::ClaudeCodeProvider;
271    use ironflow_engine::context::WorkflowContext;
272    use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
273    use ironflow_engine::prelude::CronSchedule;
274    use ironflow_store::memory::InMemoryStore;
275    use std::sync::Arc;
276    use uuid::Uuid;
277
278    use super::*;
279
280    struct NamedScheduled {
281        wf_name: &'static str,
282        cron: CronSchedule,
283    }
284    impl WorkflowHandler for NamedScheduled {
285        fn name(&self) -> &str {
286            self.wf_name
287        }
288        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
289            Box::pin(async { Ok(()) })
290        }
291        fn schedule(&self) -> Option<&CronSchedule> {
292            Some(&self.cron)
293        }
294    }
295
296    #[tokio::test]
297    async fn creates_missing_handler_schedules() {
298        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
299        let provider = Arc::new(ClaudeCodeProvider::new());
300        let mut engine = ironflow_engine::engine::Engine::new(store.clone(), provider);
301        engine
302            .register(NamedScheduled {
303                wf_name: "nightly-cleanup",
304                cron: CronSchedule::new("0 0 * * *").unwrap(),
305            })
306            .expect("register");
307
308        sync_handler_schedules(&engine, store.as_ref())
309            .await
310            .expect("sync");
311
312        let page = store.list_schedules(1, 10).await.expect("list");
313        assert_eq!(page.items.len(), 1);
314        assert_eq!(page.items[0].workflow_name, "nightly-cleanup");
315        assert_eq!(page.items[0].source, ScheduleSource::Handler);
316    }
317
318    #[tokio::test]
319    async fn skips_existing_handler_schedule() {
320        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
321        store
322            .create_schedule(NewSchedule {
323                workflow_name: "deploy".to_string(),
324                cron_expression: "*/5 * * * *".to_string(),
325                inputs: json!({"env": "prod"}),
326                source: ScheduleSource::Handler,
327                priority: 0,
328                created_by_user_id: None,
329                next_trigger_at: None,
330            })
331            .await
332            .expect("seed");
333
334        let provider = Arc::new(ClaudeCodeProvider::new());
335        let mut engine = ironflow_engine::engine::Engine::new(store.clone(), provider);
336        engine
337            .register(NamedScheduled {
338                wf_name: "deploy",
339                cron: CronSchedule::new("*/5 * * * *").unwrap(),
340            })
341            .expect("register");
342
343        sync_handler_schedules(&engine, store.as_ref())
344            .await
345            .expect("sync");
346
347        let page = store.list_schedules(1, 10).await.expect("list");
348        assert_eq!(page.items.len(), 1);
349        assert_eq!(page.items[0].cron_expression, "*/5 * * * *");
350    }
351
352    #[tokio::test]
353    async fn removes_orphan_handler_schedules() {
354        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
355        store
356            .create_schedule(NewSchedule {
357                workflow_name: "removed-workflow".to_string(),
358                cron_expression: "0 0 * * *".to_string(),
359                inputs: json!({}),
360                source: ScheduleSource::Handler,
361                priority: 0,
362                created_by_user_id: None,
363                next_trigger_at: None,
364            })
365            .await
366            .expect("seed orphan");
367
368        // API schedule must NOT be removed.
369        store
370            .create_schedule(NewSchedule {
371                workflow_name: "user-created".to_string(),
372                cron_expression: "0 12 * * *".to_string(),
373                inputs: json!({}),
374                source: ScheduleSource::Api,
375                priority: 0,
376                created_by_user_id: Some(Uuid::now_v7()),
377                next_trigger_at: None,
378            })
379            .await
380            .expect("seed api");
381
382        let provider = Arc::new(ClaudeCodeProvider::new());
383        let engine = ironflow_engine::engine::Engine::new(store.clone(), provider);
384
385        sync_handler_schedules(&engine, store.as_ref())
386            .await
387            .expect("sync");
388
389        let page = store.list_schedules(1, 10).await.expect("list");
390        assert_eq!(page.items.len(), 1);
391        assert_eq!(page.items[0].workflow_name, "user-created");
392        assert_eq!(page.items[0].source, ScheduleSource::Api);
393    }
394
395    #[tokio::test]
396    async fn updates_changed_cron() {
397        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
398        store
399            .create_schedule(NewSchedule {
400                workflow_name: "deploy".to_string(),
401                cron_expression: "0 0 * * *".to_string(),
402                inputs: json!({}),
403                source: ScheduleSource::Handler,
404                priority: 0,
405                created_by_user_id: None,
406                next_trigger_at: None,
407            })
408            .await
409            .expect("seed");
410
411        let provider = Arc::new(ClaudeCodeProvider::new());
412        let mut engine = ironflow_engine::engine::Engine::new(store.clone(), provider);
413        engine
414            .register(NamedScheduled {
415                wf_name: "deploy",
416                cron: CronSchedule::new("0 */6 * * *").unwrap(),
417            })
418            .expect("register");
419
420        sync_handler_schedules(&engine, store.as_ref())
421            .await
422            .expect("sync");
423
424        let page = store.list_schedules(1, 10).await.expect("list");
425        assert_eq!(page.items.len(), 1);
426        assert_eq!(page.items[0].cron_expression, "0 */6 * * *");
427    }
428
429    fn engine_with(store: &Arc<dyn Store>, wf_name: &'static str, cron: &str) -> Engine {
430        let provider = Arc::new(ClaudeCodeProvider::new());
431        let mut engine = Engine::new(store.clone(), provider);
432        engine
433            .register(NamedScheduled {
434                wf_name,
435                cron: CronSchedule::new(cron).unwrap(),
436            })
437            .expect("register");
438        engine
439    }
440
441    async fn seed(store: &Arc<dyn Store>, workflow: &str, cron: &str) -> Schedule {
442        store
443            .create_schedule(NewSchedule {
444                workflow_name: workflow.to_string(),
445                cron_expression: cron.to_string(),
446                inputs: json!({}),
447                source: ScheduleSource::Api,
448                priority: 0,
449                created_by_user_id: None,
450                next_trigger_at: None,
451            })
452            .await
453            .expect("seed")
454    }
455
456    async fn reload(store: &Arc<dyn Store>, id: Uuid) -> Schedule {
457        store
458            .find_schedule_by_id(id)
459            .await
460            .expect("find")
461            .expect("exists")
462    }
463
464    /// #173: a handler cron with no computable next occurrence used to be
465    /// synced as an active schedule without next trigger, never fired.
466    #[tokio::test]
467    async fn sync_disables_a_handler_schedule_whose_next_trigger_cannot_be_computed() {
468        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
469        // February 30th never happens.
470        let engine = engine_with(&store, "leap", "0 0 30 2 *");
471
472        sync_handler_schedules(&engine, store.as_ref())
473            .await
474            .expect("sync");
475
476        let page = store.list_schedules(1, 10).await.expect("list");
477        assert_eq!(page.items.len(), 1);
478        let s = &page.items[0];
479        assert!(
480            !s.is_active(),
481            "active schedule without next trigger: {s:?}"
482        );
483        assert!(s.next_trigger_at.is_none());
484        let error = s.last_error.as_deref().expect("disable reason stored");
485        assert!(error.contains("cannot compute next trigger"), "{error}");
486    }
487
488    #[tokio::test]
489    async fn sync_reenables_a_handler_schedule_disabled_on_error_once_its_cron_works() {
490        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
491        sync_handler_schedules(&engine_with(&store, "leap", "0 0 30 2 *"), store.as_ref())
492            .await
493            .expect("first sync");
494
495        sync_handler_schedules(&engine_with(&store, "leap", "0 0 1 3 *"), store.as_ref())
496            .await
497            .expect("second sync");
498
499        let page = store.list_schedules(1, 10).await.expect("list");
500        let s = &page.items[0];
501        assert!(s.is_active());
502        assert!(s.next_trigger_at.expect("next trigger") > Utc::now());
503        assert!(s.last_error.is_none());
504    }
505
506    #[tokio::test]
507    async fn sync_keeps_a_user_paused_handler_schedule_paused_on_cron_change() {
508        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
509        sync_handler_schedules(&engine_with(&store, "deploy", "0 0 * * *"), store.as_ref())
510            .await
511            .expect("first sync");
512        let id = store.list_schedules(1, 10).await.expect("list").items[0].id;
513        store
514            .update_schedule(
515                id,
516                ScheduleUpdate {
517                    disabled_at: Some(Some(Utc::now())),
518                    ..Default::default()
519                },
520            )
521            .await
522            .expect("pause");
523
524        sync_handler_schedules(
525            &engine_with(&store, "deploy", "0 */6 * * *"),
526            store.as_ref(),
527        )
528        .await
529        .expect("second sync");
530
531        let s = reload(&store, id).await;
532        assert!(!s.is_active(), "a user pause survives a cron change");
533        assert!(s.next_trigger_at.is_some());
534    }
535
536    #[tokio::test]
537    async fn repair_gives_a_next_trigger_to_an_active_schedule_without_one() {
538        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
539        let broken = seed(&store, "hourly", "0 * * * *").await;
540        let paused = seed(&store, "paused", "0 * * * *").await;
541        store
542            .update_schedule(
543                paused.id,
544                ScheduleUpdate {
545                    disabled_at: Some(Some(Utc::now())),
546                    ..Default::default()
547                },
548            )
549            .await
550            .expect("pause");
551
552        let repaired = repair_unscheduled_schedules(store.as_ref())
553            .await
554            .expect("repair");
555
556        assert_eq!(repaired, 1);
557        let s = reload(&store, broken.id).await;
558        assert!(s.is_active());
559        assert!(s.next_trigger_at.expect("next trigger") > Utc::now());
560        assert!(s.last_error.is_none());
561        // A paused schedule needs no next trigger: left alone.
562        let p = reload(&store, paused.id).await;
563        assert!(!p.is_active());
564        assert!(p.next_trigger_at.is_none());
565    }
566
567    #[tokio::test]
568    async fn repair_disables_a_schedule_whose_next_trigger_cannot_be_computed() {
569        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
570        let broken = seed(&store, "leap", "0 0 30 2 *").await;
571
572        let repaired = repair_unscheduled_schedules(store.as_ref())
573            .await
574            .expect("repair");
575
576        assert_eq!(repaired, 1);
577        let s = reload(&store, broken.id).await;
578        assert!(!s.is_active());
579        let error = s.last_error.expect("disable reason stored");
580        assert!(error.contains("cannot compute next trigger"), "{error}");
581    }
582
583    #[tokio::test]
584    async fn repair_reads_every_page() {
585        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
586        let count = REPAIR_PAGE_SIZE as usize * 2 + 3;
587        for i in 0..count {
588            seed(&store, &format!("wf-{i}"), "0 * * * *").await;
589        }
590
591        let repaired = repair_unscheduled_schedules(store.as_ref())
592            .await
593            .expect("repair");
594
595        assert_eq!(repaired, count);
596        let left = store
597            .list_schedules(1, 1000)
598            .await
599            .expect("list")
600            .items
601            .into_iter()
602            .filter(|s| s.is_active() && s.next_trigger_at.is_none())
603            .count();
604        assert_eq!(left, 0);
605    }
606
607    #[tokio::test]
608    async fn repair_with_nothing_to_repair_changes_nothing() {
609        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
610        let repaired = repair_unscheduled_schedules(store.as_ref())
611            .await
612            .expect("repair");
613        assert_eq!(repaired, 0);
614    }
615
616    struct PriorityScheduled {
617        priority: i16,
618        cron: CronSchedule,
619    }
620    impl WorkflowHandler for PriorityScheduled {
621        fn name(&self) -> &str {
622            "urgent-report"
623        }
624        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
625            Box::pin(async { Ok(()) })
626        }
627        fn schedule(&self) -> Option<&CronSchedule> {
628            Some(&self.cron)
629        }
630        fn priority(&self) -> i16 {
631            self.priority
632        }
633    }
634
635    fn engine_with_priority(store: &Arc<dyn Store>, priority: i16) -> Engine {
636        let provider = Arc::new(ClaudeCodeProvider::new());
637        let mut engine = Engine::new(store.clone(), provider);
638        engine
639            .register(PriorityScheduled {
640                priority,
641                cron: CronSchedule::new("0 0 * * *").unwrap(),
642            })
643            .expect("register");
644        engine
645    }
646
647    async fn only_schedule(store: &Arc<dyn Store>) -> Schedule {
648        let page = store.list_schedules(1, 10).await.expect("list");
649        assert_eq!(page.items.len(), 1);
650        page.items.into_iter().next().expect("one schedule")
651    }
652
653    #[tokio::test]
654    async fn sync_creates_a_handler_schedule_with_the_handler_priority() {
655        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
656        let engine = engine_with_priority(&store, 40);
657
658        sync_handler_schedules(&engine, store.as_ref())
659            .await
660            .expect("sync");
661
662        let schedule = only_schedule(&store).await;
663        assert_eq!(schedule.source, ScheduleSource::Handler);
664        assert_eq!(schedule.priority, 40);
665        assert_eq!(schedule.new_run(None).priority, 40);
666    }
667
668    #[tokio::test]
669    async fn sync_updates_the_priority_when_the_handler_priority_changed() {
670        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
671        sync_handler_schedules(&engine_with_priority(&store, 0), store.as_ref())
672            .await
673            .expect("first sync");
674        let before = only_schedule(&store).await;
675        assert_eq!(before.priority, 0);
676
677        sync_handler_schedules(&engine_with_priority(&store, -30), store.as_ref())
678            .await
679            .expect("second sync");
680
681        let after = only_schedule(&store).await;
682        assert_eq!(after.id, before.id);
683        assert_eq!(after.priority, -30);
684        assert_eq!(after.cron_expression, before.cron_expression);
685        assert_eq!(after.next_trigger_at, before.next_trigger_at);
686    }
687
688    #[tokio::test]
689    async fn sync_clamps_an_out_of_range_handler_priority() {
690        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
691        sync_handler_schedules(&engine_with_priority(&store, 500), store.as_ref())
692            .await
693            .expect("sync high");
694        assert_eq!(only_schedule(&store).await.priority, MAX_PRIORITY);
695
696        sync_handler_schedules(&engine_with_priority(&store, -500), store.as_ref())
697            .await
698            .expect("sync low");
699        assert_eq!(only_schedule(&store).await.priority, MIN_PRIORITY);
700    }
701}
ironflow_api/schedule_sync.rs

ironflow_api/
schedule_sync.rs
