graphrefly/work_queue/
readiness.rs

1//! workQueue convergence onto scheduled readiness (B95 over D314-D318/D424-D433).
2//!
3//! This module translates existing Rust workQueue delayed eligibility records
4//! into shared scheduled-readiness facts and consumes readiness back into
5//! workQueue-owned candidate/status material. It does not append queue lifecycle
6//! records and does not claim, expire, cancel, complete, or fail work.
7
8use std::cell::RefCell;
9use std::collections::{BTreeMap, BTreeSet};
10use std::rc::Rc;
11
12use crate::ctx::Ctx;
13use crate::graph::{Graph, GraphNodeOpts};
14use crate::identity::{canonical_tuple_key, compound_tuple_key};
15use crate::json::JsonValue;
16use crate::messaging::DataIssue;
17use crate::node::{Node, NodeOpts};
18use crate::scheduled_readiness::{
19    readiness_issue, ScheduledReadinessAuditRecord, ScheduledReadinessOverdue,
20    ScheduledReadinessReady, ScheduledReadinessRequested, SourceRef,
21};
22use crate::work_queue::{WorkQueueCommand, WorkQueueDerivedState, WorkQueueRecord};
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25/// `WorkQueueReadinessScheduleKind` variants.
26pub enum WorkQueueReadinessScheduleKind {
27    /// `AdmissionDelay` variant.
28    AdmissionDelay,
29    /// `WorkScheduled` variant.
30    WorkScheduled,
31    /// `RetryScheduled` variant.
32    RetryScheduled,
33    /// `LeaseExpiration` variant.
34    LeaseExpiration,
35}
36
37impl WorkQueueReadinessScheduleKind {
38    fn as_str(self) -> &'static str {
39        match self {
40            Self::AdmissionDelay => "admission-delay",
41            Self::WorkScheduled => "work-scheduled",
42            Self::RetryScheduled => "retry-scheduled",
43            Self::LeaseExpiration => "lease-expiration",
44        }
45    }
46
47    fn from_str(value: &str) -> Option<Self> {
48        match value {
49            "admission-delay" => Some(Self::AdmissionDelay),
50            "work-scheduled" => Some(Self::WorkScheduled),
51            "retry-scheduled" => Some(Self::RetryScheduled),
52            "lease-expiration" => Some(Self::LeaseExpiration),
53            _ => None,
54        }
55    }
56}
57
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59/// `WorkQueueReadinessStatusState` variants.
60pub enum WorkQueueReadinessStatusState {
61    /// `Translated` variant.
62    Translated,
63    /// `Candidate` variant.
64    Candidate,
65    /// `Ignored` variant.
66    Ignored,
67    /// `Overdue` variant.
68    Overdue,
69    /// `Issue` variant.
70    Issue,
71}
72
73#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74/// `WorkQueueReadinessCandidateKind` variants.
75pub enum WorkQueueReadinessCandidateKind {
76    /// `ClaimEligible` variant.
77    ClaimEligible,
78    /// `LeaseExpirationEligible` variant.
79    LeaseExpirationEligible,
80}
81
82impl WorkQueueReadinessCandidateKind {
83    fn as_str(self) -> &'static str {
84        match self {
85            Self::ClaimEligible => "claim-eligible",
86            Self::LeaseExpirationEligible => "lease-expiration-eligible",
87        }
88    }
89}
90
91#[derive(Debug, Clone, PartialEq)]
92/// `WorkQueueReadinessStatus` data container.
93pub struct WorkQueueReadinessStatus {
94    /// `status_id` field for status id.
95    pub status_id: String,
96    /// `queue_id` field for queue id.
97    pub queue_id: String,
98    /// `work_id` field for work id.
99    pub work_id: Option<String>,
100    /// `schedule_id` field for schedule id.
101    pub schedule_id: Option<String>,
102    /// `state` field for state.
103    pub state: WorkQueueReadinessStatusState,
104    /// `schedule_kind` field for schedule kind.
105    pub schedule_kind: Option<WorkQueueReadinessScheduleKind>,
106    /// `ready_at_ms` field for ready at ms.
107    pub ready_at_ms: Option<u64>,
108    /// `now_ms` field for now ms.
109    pub now_ms: Option<u64>,
110    /// `issue_code` field for issue code.
111    pub issue_code: Option<String>,
112    /// `details` field for details.
113    pub details: Option<String>,
114}
115
116#[derive(Debug, Clone, PartialEq)]
117/// `WorkQueueReadinessCandidate` data container.
118pub struct WorkQueueReadinessCandidate {
119    /// `candidate_id` field for candidate id.
120    pub candidate_id: String,
121    /// `queue_id` field for queue id.
122    pub queue_id: String,
123    /// `work_id` field for work id.
124    pub work_id: String,
125    /// `schedule_id` field for schedule id.
126    pub schedule_id: String,
127    /// `candidate_kind` field for candidate kind.
128    pub candidate_kind: WorkQueueReadinessCandidateKind,
129    /// `schedule_kind` field for schedule kind.
130    pub schedule_kind: WorkQueueReadinessScheduleKind,
131    /// `ready_at_ms` field for ready at ms.
132    pub ready_at_ms: u64,
133    /// `now_ms` field for now ms.
134    pub now_ms: u64,
135    /// `lease_id` field for lease id.
136    pub lease_id: Option<String>,
137    /// `attempt` field for attempt.
138    pub attempt: Option<u32>,
139    /// `worker_id` field for worker id.
140    pub worker_id: Option<String>,
141    /// `source_refs` field for source refs.
142    pub source_refs: Vec<SourceRef>,
143}
144
145#[derive(Debug, Clone, PartialEq, Default)]
146/// `WorkQueueReadinessViews` data container.
147pub struct WorkQueueReadinessViews {
148    /// `schedules_by_id` field for schedules by id.
149    pub schedules_by_id: BTreeMap<String, ScheduledReadinessRequested>,
150    /// `candidates_by_id` field for candidates by id.
151    pub candidates_by_id: BTreeMap<String, WorkQueueReadinessCandidate>,
152    /// `status_by_id` field for status by id.
153    pub status_by_id: BTreeMap<String, WorkQueueReadinessStatus>,
154}
155
156#[derive(Clone)]
157/// `WorkQueueScheduledReadinessOptions` data container.
158pub struct WorkQueueScheduledReadinessOptions<T> {
159    /// `name` field for name.
160    pub name: Option<String>,
161    /// `records` field for records.
162    pub records: Vec<Node<WorkQueueRecord<T>>>,
163}
164
165impl<T> WorkQueueScheduledReadinessOptions<T> {
166    /// Creates or computes `new`.
167    pub fn new(records: Vec<Node<WorkQueueRecord<T>>>) -> Self {
168        Self {
169            name: None,
170            records,
171        }
172    }
173
174    /// Updates or reads `named`.
175    pub fn named(mut self, name: impl Into<String>) -> Self {
176        self.name = Some(name.into());
177        self
178    }
179}
180
181#[derive(Clone)]
182/// `WorkQueueReadinessHandoffOptions` data container.
183pub struct WorkQueueReadinessHandoffOptions<T> {
184    /// `name` field for name.
185    pub name: Option<String>,
186    /// `records` field for records.
187    pub records: Vec<Node<WorkQueueRecord<T>>>,
188    /// `ready` field for ready.
189    pub ready: Vec<Node<ScheduledReadinessReady>>,
190    /// `overdue` field for overdue.
191    pub overdue: Vec<Node<ScheduledReadinessOverdue>>,
192}
193
194impl<T> WorkQueueReadinessHandoffOptions<T> {
195    /// Creates or computes `new`.
196    pub fn new(
197        records: Vec<Node<WorkQueueRecord<T>>>,
198        ready: Vec<Node<ScheduledReadinessReady>>,
199    ) -> Self {
200        Self {
201            name: None,
202            records,
203            ready,
204            overdue: Vec::new(),
205        }
206    }
207
208    /// Updates or reads `named`.
209    pub fn named(mut self, name: impl Into<String>) -> Self {
210        self.name = Some(name.into());
211        self
212    }
213
214    /// Updates or reads `with_overdue`.
215    pub fn with_overdue(mut self, overdue: Vec<Node<ScheduledReadinessOverdue>>) -> Self {
216        self.overdue = overdue;
217        self
218    }
219}
220
221#[derive(Clone)]
222/// `WorkQueueScheduledReadinessBundle` data container.
223pub struct WorkQueueScheduledReadinessBundle {
224    /// `readiness_schedules` field for readiness schedules.
225    pub readiness_schedules: Node<ScheduledReadinessRequested>,
226    /// `status` field for status.
227    pub status: Node<WorkQueueReadinessStatus>,
228    /// `issues` field for issues.
229    pub issues: Node<DataIssue>,
230    /// `audit` field for audit.
231    pub audit: Node<ScheduledReadinessAuditRecord>,
232    /// `views` field for views.
233    pub views: Node<WorkQueueReadinessViews>,
234}
235
236#[derive(Clone)]
237/// `WorkQueueReadinessHandoffBundle` data container.
238pub struct WorkQueueReadinessHandoffBundle {
239    /// `candidates` field for candidates.
240    pub candidates: Node<WorkQueueReadinessCandidate>,
241    /// `status` field for status.
242    pub status: Node<WorkQueueReadinessStatus>,
243    /// `issues` field for issues.
244    pub issues: Node<DataIssue>,
245    /// `audit` field for audit.
246    pub audit: Node<ScheduledReadinessAuditRecord>,
247    /// `views` field for views.
248    pub views: Node<WorkQueueReadinessViews>,
249}
250
251#[derive(Clone)]
252/// `WorkQueueLeaseExpirationCommandProjectorOptions` data container.
253pub struct WorkQueueLeaseExpirationCommandProjectorOptions {
254    /// `name` field for name.
255    pub name: Option<String>,
256    /// `candidates` field for candidates.
257    pub candidates: Vec<Node<WorkQueueReadinessCandidate>>,
258    /// `command_prefix` field for command prefix.
259    pub command_prefix: Option<String>,
260}
261
262impl WorkQueueLeaseExpirationCommandProjectorOptions {
263    /// Creates or computes `new`.
264    pub fn new(candidates: Vec<Node<WorkQueueReadinessCandidate>>) -> Self {
265        Self {
266            name: None,
267            candidates,
268            command_prefix: None,
269        }
270    }
271
272    /// Updates or reads `named`.
273    pub fn named(mut self, name: impl Into<String>) -> Self {
274        self.name = Some(name.into());
275        self
276    }
277
278    /// Updates or reads `command_prefix`.
279    pub fn command_prefix(mut self, command_prefix: impl Into<String>) -> Self {
280        self.command_prefix = Some(command_prefix.into());
281        self
282    }
283}
284
285#[derive(Clone)]
286enum WorkQueueReadinessFact {
287    Schedule(ScheduledReadinessRequested),
288    Candidate(WorkQueueReadinessCandidate),
289    Status(WorkQueueReadinessStatus),
290    Issue(DataIssue),
291    Audit(ScheduledReadinessAuditRecord),
292    Views(WorkQueueReadinessViews),
293}
294
295#[derive(Default)]
296struct TranslatorState {
297    schedules_by_id: BTreeMap<String, ScheduledReadinessRequested>,
298    status_by_id: BTreeMap<String, WorkQueueReadinessStatus>,
299    emitted: BTreeSet<String>,
300    audit_seq: u64,
301}
302
303#[derive(Debug, Clone)]
304struct QueueWorkState {
305    queue_id: String,
306    state: WorkQueueDerivedState,
307    effective_ready_at_ms: Option<u64>,
308    effective_schedule_kind: Option<WorkQueueReadinessScheduleKind>,
309    lease_id: Option<String>,
310    attempt: Option<u32>,
311    worker_id: Option<String>,
312    lease_expires_at_ms: Option<u64>,
313}
314
315#[derive(Default)]
316struct HandoffState {
317    works_by_id: BTreeMap<(String, String), QueueWorkState>,
318    ready_by_id: BTreeMap<String, ScheduledReadinessReady>,
319    candidates_by_id: BTreeMap<String, WorkQueueReadinessCandidate>,
320    status_by_id: BTreeMap<String, WorkQueueReadinessStatus>,
321    emitted: BTreeSet<String>,
322    audit_seq: u64,
323}
324
325/// Creates or computes `work_queue_scheduled_readiness_projector`.
326pub fn work_queue_scheduled_readiness_projector<T: Clone + 'static>(
327    graph: &Graph,
328    opts: WorkQueueScheduledReadinessOptions<T>,
329) -> WorkQueueScheduledReadinessBundle {
330    let name = opts
331        .name
332        .clone()
333        .unwrap_or_else(|| "workQueueScheduledReadiness".to_owned());
334    let deps = opts.records.iter().map(Node::erased).collect::<Vec<_>>();
335    let dep_count = opts.records.len();
336    let state = Rc::new(RefCell::new(TranslatorState::default()));
337    let runtime = graph.node_opts::<WorkQueueReadinessFact, _>(
338        deps,
339        {
340            let state = state.clone();
341            move |ctx| {
342                let mut state = state.borrow_mut();
343                for index in 0..dep_count {
344                    for record in ctx.batch::<WorkQueueRecord<T>>(index) {
345                        if let Some(schedule) = schedule_from_record(record.as_ref()) {
346                            emit_translated_schedule(ctx, &mut state, schedule);
347                        }
348                    }
349                }
350                emit_fact(
351                    ctx,
352                    WorkQueueReadinessFact::Views(WorkQueueReadinessViews {
353                        schedules_by_id: state.schedules_by_id.clone(),
354                        candidates_by_id: BTreeMap::new(),
355                        status_by_id: state.status_by_id.clone(),
356                    }),
357                );
358            }
359        },
360        {
361            let mut node_opts = node_opts(
362                format!("{name}/runtime"),
363                "workQueueScheduledReadinessProjector",
364            );
365            node_opts.node.partial = true;
366            node_opts
367        },
368    );
369    WorkQueueScheduledReadinessBundle {
370        readiness_schedules: project_fact(
371            graph,
372            &runtime,
373            format!("{name}/schedules"),
374            "workQueueReadinessSchedules",
375            |fact| match fact {
376                WorkQueueReadinessFact::Schedule(value) => Some(value.clone()),
377                _ => None,
378            },
379        ),
380        status: project_fact(
381            graph,
382            &runtime,
383            format!("{name}/status"),
384            "workQueueReadinessStatus",
385            |fact| match fact {
386                WorkQueueReadinessFact::Status(value) => Some(value.clone()),
387                _ => None,
388            },
389        ),
390        issues: project_fact(
391            graph,
392            &runtime,
393            format!("{name}/issues"),
394            "workQueueReadinessIssues",
395            |fact| match fact {
396                WorkQueueReadinessFact::Issue(value) => Some(value.clone()),
397                _ => None,
398            },
399        ),
400        audit: project_fact(
401            graph,
402            &runtime,
403            format!("{name}/audit"),
404            "workQueueReadinessAudit",
405            |fact| match fact {
406                WorkQueueReadinessFact::Audit(value) => Some(value.clone()),
407                _ => None,
408            },
409        ),
410        views: project_fact(
411            graph,
412            &runtime,
413            format!("{name}/views"),
414            "workQueueReadinessViews",
415            |fact| match fact {
416                WorkQueueReadinessFact::Views(value) => Some(value.clone()),
417                _ => None,
418            },
419        ),
420    }
421}
422
423/// Creates or computes `work_queue_readiness_handoff_projector`.
424pub fn work_queue_readiness_handoff_projector<T: Clone + 'static>(
425    graph: &Graph,
426    opts: WorkQueueReadinessHandoffOptions<T>,
427) -> WorkQueueReadinessHandoffBundle {
428    let name = opts
429        .name
430        .clone()
431        .unwrap_or_else(|| "workQueueReadinessHandoff".to_owned());
432    let record_count = opts.records.len();
433    let ready_start = record_count;
434    let overdue_start = ready_start + opts.ready.len();
435    let mut deps = Vec::with_capacity(opts.records.len() + opts.ready.len() + opts.overdue.len());
436    deps.extend(opts.records.iter().map(Node::erased));
437    deps.extend(opts.ready.iter().map(Node::erased));
438    deps.extend(opts.overdue.iter().map(Node::erased));
439    let state = Rc::new(RefCell::new(HandoffState::default()));
440    let runtime = graph.node_opts::<WorkQueueReadinessFact, _>(
441        deps,
442        {
443            let state = state.clone();
444            move |ctx| {
445                let mut state = state.borrow_mut();
446                for index in 0..record_count {
447                    for record in ctx.batch::<WorkQueueRecord<T>>(index) {
448                        apply_record(&mut state, record.as_ref());
449                    }
450                }
451                for ready in state.ready_by_id.values().cloned().collect::<Vec<_>>() {
452                    handoff_ready_inner(ctx, &mut state, &ready);
453                }
454                for index in ready_start..overdue_start {
455                    for ready in ctx.batch::<ScheduledReadinessReady>(index) {
456                        handoff_ready(ctx, &mut state, ready.as_ref());
457                    }
458                }
459                for index in overdue_start..(overdue_start + opts.overdue.len()) {
460                    for overdue in ctx.batch::<ScheduledReadinessOverdue>(index) {
461                        handoff_overdue(ctx, &mut state, overdue.as_ref());
462                    }
463                }
464                emit_fact(
465                    ctx,
466                    WorkQueueReadinessFact::Views(WorkQueueReadinessViews {
467                        schedules_by_id: BTreeMap::new(),
468                        candidates_by_id: state.candidates_by_id.clone(),
469                        status_by_id: state.status_by_id.clone(),
470                    }),
471                );
472            }
473        },
474        {
475            let mut node_opts = node_opts(
476                format!("{name}/runtime"),
477                "workQueueReadinessHandoffProjector",
478            );
479            node_opts.node.partial = true;
480            node_opts
481        },
482    );
483    WorkQueueReadinessHandoffBundle {
484        candidates: project_fact(
485            graph,
486            &runtime,
487            format!("{name}/candidates"),
488            "workQueueReadinessCandidates",
489            |fact| match fact {
490                WorkQueueReadinessFact::Candidate(value) => Some(value.clone()),
491                _ => None,
492            },
493        ),
494        status: project_fact(
495            graph,
496            &runtime,
497            format!("{name}/status"),
498            "workQueueReadinessStatus",
499            |fact| match fact {
500                WorkQueueReadinessFact::Status(value) => Some(value.clone()),
501                _ => None,
502            },
503        ),
504        issues: project_fact(
505            graph,
506            &runtime,
507            format!("{name}/issues"),
508            "workQueueReadinessIssues",
509            |fact| match fact {
510                WorkQueueReadinessFact::Issue(value) => Some(value.clone()),
511                _ => None,
512            },
513        ),
514        audit: project_fact(
515            graph,
516            &runtime,
517            format!("{name}/audit"),
518            "workQueueReadinessAudit",
519            |fact| match fact {
520                WorkQueueReadinessFact::Audit(value) => Some(value.clone()),
521                _ => None,
522            },
523        ),
524        views: project_fact(
525            graph,
526            &runtime,
527            format!("{name}/views"),
528            "workQueueReadinessViews",
529            |fact| match fact {
530                WorkQueueReadinessFact::Views(value) => Some(value.clone()),
531                _ => None,
532            },
533        ),
534    }
535}
536
537/// Creates or computes `work_queue_lease_expiration_command_projector`.
538pub fn work_queue_lease_expiration_command_projector<T: Clone + 'static>(
539    graph: &Graph,
540    opts: WorkQueueLeaseExpirationCommandProjectorOptions,
541) -> Node<WorkQueueCommand<T>> {
542    let name = opts
543        .name
544        .clone()
545        .unwrap_or_else(|| "workQueueLeaseExpirationCommands".to_owned());
546    let prefix = opts
547        .command_prefix
548        .clone()
549        .unwrap_or_else(|| "readiness-expire".to_owned());
550    let dep_count = opts.candidates.len();
551    graph.node_opts::<WorkQueueCommand<T>, _>(
552        opts.candidates.iter().map(Node::erased).collect(),
553        move |ctx| {
554            for index in 0..dep_count {
555                for candidate in ctx.batch::<WorkQueueReadinessCandidate>(index) {
556                    if let Some(command) =
557                        work_queue_lease_expiration_command::<T>(candidate.as_ref(), &prefix)
558                    {
559                        ctx.emit(command);
560                    }
561                }
562            }
563        },
564        node_opts(name, "workQueueLeaseExpirationCommandProjector"),
565    )
566}
567
568/// Creates or computes `work_queue_lease_expiration_command`.
569pub fn work_queue_lease_expiration_command<T>(
570    candidate: &WorkQueueReadinessCandidate,
571    command_prefix: &str,
572) -> Option<WorkQueueCommand<T>> {
573    if candidate.candidate_kind != WorkQueueReadinessCandidateKind::LeaseExpirationEligible {
574        return None;
575    }
576    Some(WorkQueueCommand::ExpireLeases {
577        command_id: compound_tuple_key(command_prefix, &[&candidate.candidate_id]),
578        queue_id: Some(candidate.queue_id.clone()),
579        idempotency_key: Some(compound_tuple_key(
580            command_prefix,
581            &[&candidate.candidate_id],
582        )),
583        work_ids: vec![candidate.work_id.clone()],
584        limit: Some(1),
585        now_ms: Some(candidate.now_ms),
586    })
587}
588
589fn schedule_from_record<T>(record: &WorkQueueRecord<T>) -> Option<ScheduledReadinessRequested> {
590    match record {
591        WorkQueueRecord::WorkAdmitted {
592            record_seq,
593            queue_id,
594            work_id,
595            not_before_ms: Some(ready_at_ms),
596            deadline_ms,
597            ..
598        } => Some(readiness_schedule(ReadinessScheduleInput {
599            queue_id,
600            work_id,
601            kind: WorkQueueReadinessScheduleKind::AdmissionDelay,
602            schedule_id: compound_tuple_key(
603                "work-queue-admission-readiness",
604                &[queue_id, work_id, &record_seq.to_string()],
605            ),
606            ready_at_ms: *ready_at_ms,
607            deadline_ms: *deadline_ms,
608            record_seq: *record_seq,
609            lease: None,
610        })),
611        WorkQueueRecord::WorkScheduled {
612            record_seq,
613            queue_id,
614            work_id,
615            command_id,
616            schedule_id,
617            not_before_ms,
618            deadline_ms,
619            ..
620        } => Some(readiness_schedule(ReadinessScheduleInput {
621            queue_id,
622            work_id,
623            kind: WorkQueueReadinessScheduleKind::WorkScheduled,
624            schedule_id: compound_tuple_key(
625                "work-queue-scheduled-readiness",
626                &[
627                    queue_id,
628                    work_id,
629                    schedule_id.as_deref().unwrap_or(command_id),
630                    &record_seq.to_string(),
631                ],
632            ),
633            ready_at_ms: *not_before_ms,
634            deadline_ms: *deadline_ms,
635            record_seq: *record_seq,
636            lease: None,
637        })),
638        WorkQueueRecord::RetryScheduled {
639            record_seq,
640            queue_id,
641            work_id,
642            command_id,
643            retry_at_ms,
644            ..
645        } => Some(readiness_schedule(ReadinessScheduleInput {
646            queue_id,
647            work_id,
648            kind: WorkQueueReadinessScheduleKind::RetryScheduled,
649            schedule_id: compound_tuple_key(
650                "work-queue-retry-readiness",
651                &[queue_id, work_id, command_id, &record_seq.to_string()],
652            ),
653            ready_at_ms: *retry_at_ms,
654            deadline_ms: None,
655            record_seq: *record_seq,
656            lease: None,
657        })),
658        WorkQueueRecord::WorkClaimed {
659            record_seq,
660            queue_id,
661            work_id,
662            lease_id,
663            attempt,
664            worker_id,
665            lease_expires_at_ms,
666            ..
667        }
668        | WorkQueueRecord::LeaseRenewed {
669            record_seq,
670            queue_id,
671            work_id,
672            lease_id,
673            attempt,
674            worker_id,
675            lease_expires_at_ms,
676            ..
677        } => Some(readiness_schedule(ReadinessScheduleInput {
678            queue_id,
679            work_id,
680            kind: WorkQueueReadinessScheduleKind::LeaseExpiration,
681            schedule_id: compound_tuple_key(
682                "work-queue-lease-readiness",
683                &[
684                    queue_id,
685                    work_id,
686                    lease_id,
687                    &attempt.to_string(),
688                    &record_seq.to_string(),
689                ],
690            ),
691            ready_at_ms: *lease_expires_at_ms,
692            deadline_ms: None,
693            record_seq: *record_seq,
694            lease: Some((lease_id.clone(), *attempt, worker_id.clone())),
695        })),
696        _ => None,
697    }
698}
699
700struct ReadinessScheduleInput<'a> {
701    queue_id: &'a str,
702    work_id: &'a str,
703    kind: WorkQueueReadinessScheduleKind,
704    schedule_id: String,
705    ready_at_ms: u64,
706    deadline_ms: Option<u64>,
707    record_seq: u64,
708    lease: Option<(String, u32, String)>,
709}
710
711fn readiness_schedule(input: ReadinessScheduleInput<'_>) -> ScheduledReadinessRequested {
712    let mut metadata = BTreeMap::from([
713        (
714            "queueId".to_owned(),
715            JsonValue::from(input.queue_id.to_owned()),
716        ),
717        (
718            "workId".to_owned(),
719            JsonValue::from(input.work_id.to_owned()),
720        ),
721        (
722            "scheduleKind".to_owned(),
723            JsonValue::from(input.kind.as_str().to_owned()),
724        ),
725        ("recordSeq".to_owned(), JsonValue::from(input.record_seq)),
726    ]);
727    if let Some((lease_id, attempt, worker_id)) = input.lease {
728        metadata.insert("leaseId".to_owned(), JsonValue::from(lease_id));
729        metadata.insert("attempt".to_owned(), JsonValue::from(attempt));
730        metadata.insert("workerId".to_owned(), JsonValue::from(worker_id));
731    }
732    ScheduledReadinessRequested {
733        schedule_id: input.schedule_id.clone(),
734        subject_refs: vec![
735            SourceRef::new("work-queue", input.queue_id.to_owned()),
736            SourceRef::new("work-queue-work", input.work_id.to_owned()),
737        ],
738        ready_at_ms: input.ready_at_ms,
739        deadline_ms: input.deadline_ms,
740        reason: Some(input.kind.as_str().to_owned()),
741        policy_refs: Vec::new(),
742        source_refs: vec![SourceRef::new(
743            format!("work-queue-{}", input.kind.as_str()),
744            input.schedule_id,
745        )],
746        metadata: Some(metadata),
747    }
748}
749
750fn emit_translated_schedule(
751    ctx: &Ctx,
752    state: &mut TranslatorState,
753    schedule: ScheduledReadinessRequested,
754) {
755    let schedule_id = schedule.schedule_id.clone();
756    if state
757        .schedules_by_id
758        .insert(schedule_id.clone(), schedule.clone())
759        .is_none()
760    {
761        emit_fact(ctx, WorkQueueReadinessFact::Schedule(schedule.clone()));
762        emit_status(
763            ctx,
764            state,
765            WorkQueueReadinessStatus {
766                status_id: compound_tuple_key("work-queue-readiness-translated", &[&schedule_id]),
767                queue_id: metadata_string(&schedule, "queueId").unwrap_or_default(),
768                work_id: metadata_string(&schedule, "workId"),
769                schedule_id: Some(schedule_id),
770                state: WorkQueueReadinessStatusState::Translated,
771                schedule_kind: metadata_kind(&schedule),
772                ready_at_ms: Some(schedule.ready_at_ms),
773                now_ms: None,
774                issue_code: None,
775                details: None,
776            },
777        );
778        emit_translator_audit(ctx, state, "work-queue-readiness-translated", &schedule);
779    }
780}
781
782fn apply_record<T>(state: &mut HandoffState, record: &WorkQueueRecord<T>) {
783    match record {
784        WorkQueueRecord::WorkAdmitted {
785            queue_id,
786            work_id,
787            not_before_ms,
788            ..
789        } => {
790            prune_candidates_for_work(state, queue_id, work_id);
791            state.works_by_id.insert(
792                work_key(queue_id, work_id),
793                QueueWorkState {
794                    queue_id: queue_id.clone(),
795                    state: if not_before_ms.is_some() {
796                        WorkQueueDerivedState::Scheduled
797                    } else {
798                        WorkQueueDerivedState::Ready
799                    },
800                    effective_ready_at_ms: *not_before_ms,
801                    effective_schedule_kind: not_before_ms
802                        .map(|_| WorkQueueReadinessScheduleKind::AdmissionDelay),
803                    lease_id: None,
804                    attempt: None,
805                    worker_id: None,
806                    lease_expires_at_ms: None,
807                },
808            );
809        }
810        WorkQueueRecord::WorkScheduled {
811            queue_id,
812            work_id,
813            not_before_ms,
814            ..
815        } => set_work_delayed_state(
816            state,
817            queue_id,
818            work_id,
819            WorkQueueDerivedState::Scheduled,
820            Some(*not_before_ms),
821            Some(WorkQueueReadinessScheduleKind::WorkScheduled),
822        ),
823        WorkQueueRecord::RetryScheduled {
824            queue_id,
825            work_id,
826            retry_at_ms,
827            ..
828        } => set_work_delayed_state(
829            state,
830            queue_id,
831            work_id,
832            WorkQueueDerivedState::RetryWait,
833            Some(*retry_at_ms),
834            Some(WorkQueueReadinessScheduleKind::RetryScheduled),
835        ),
836        WorkQueueRecord::WorkClaimed {
837            queue_id,
838            work_id,
839            lease_id,
840            attempt,
841            worker_id,
842            lease_expires_at_ms,
843            ..
844        }
845        | WorkQueueRecord::LeaseRenewed {
846            queue_id,
847            work_id,
848            lease_id,
849            attempt,
850            worker_id,
851            lease_expires_at_ms,
852            ..
853        } => {
854            prune_candidates_for_work(state, queue_id, work_id);
855            state.works_by_id.insert(
856                work_key(queue_id, work_id),
857                QueueWorkState {
858                    queue_id: queue_id.clone(),
859                    state: WorkQueueDerivedState::Leased,
860                    effective_ready_at_ms: Some(*lease_expires_at_ms),
861                    effective_schedule_kind: Some(WorkQueueReadinessScheduleKind::LeaseExpiration),
862                    lease_id: Some(lease_id.clone()),
863                    attempt: Some(*attempt),
864                    worker_id: Some(worker_id.clone()),
865                    lease_expires_at_ms: Some(*lease_expires_at_ms),
866                },
867            );
868        }
869        WorkQueueRecord::WorkReleased {
870            queue_id, work_id, ..
871        }
872        | WorkQueueRecord::LeaseExpired {
873            queue_id, work_id, ..
874        } => set_work_state(state, queue_id, work_id, WorkQueueDerivedState::Ready),
875        WorkQueueRecord::WorkCompleted {
876            queue_id, work_id, ..
877        } => set_work_state(state, queue_id, work_id, WorkQueueDerivedState::Completed),
878        WorkQueueRecord::WorkCanceled {
879            queue_id, work_id, ..
880        } => set_work_state(state, queue_id, work_id, WorkQueueDerivedState::Canceled),
881        WorkQueueRecord::WorkDeadLettered {
882            queue_id, work_id, ..
883        } => set_work_state(
884            state,
885            queue_id,
886            work_id,
887            WorkQueueDerivedState::DeadLettered,
888        ),
889        WorkQueueRecord::AttemptFailed { .. }
890        | WorkQueueRecord::AttemptCompleted { .. }
891        | WorkQueueRecord::AdmissionDeduped { .. } => {}
892    }
893}
894
895fn set_work_state(
896    state: &mut HandoffState,
897    queue_id: &str,
898    work_id: &str,
899    next: WorkQueueDerivedState,
900) {
901    let terminal = matches!(
902        next,
903        WorkQueueDerivedState::Completed
904            | WorkQueueDerivedState::Canceled
905            | WorkQueueDerivedState::DeadLettered
906    );
907    if terminal {
908        prune_candidates_for_work(state, queue_id, work_id);
909    }
910    set_work_delayed_state(state, queue_id, work_id, next, None, None);
911}
912
913fn set_work_delayed_state(
914    state: &mut HandoffState,
915    queue_id: &str,
916    work_id: &str,
917    next: WorkQueueDerivedState,
918    effective_ready_at_ms: Option<u64>,
919    effective_schedule_kind: Option<WorkQueueReadinessScheduleKind>,
920) {
921    prune_candidates_for_work(state, queue_id, work_id);
922    state
923        .works_by_id
924        .entry(work_key(queue_id, work_id))
925        .and_modify(|work| {
926            work.state = next;
927            work.effective_ready_at_ms = effective_ready_at_ms;
928            work.effective_schedule_kind = effective_schedule_kind;
929            if next != WorkQueueDerivedState::Leased {
930                work.lease_id = None;
931                work.attempt = None;
932                work.worker_id = None;
933                work.lease_expires_at_ms = None;
934            }
935        })
936        .or_insert_with(|| QueueWorkState {
937            queue_id: queue_id.to_owned(),
938            state: next,
939            effective_ready_at_ms,
940            effective_schedule_kind,
941            lease_id: None,
942            attempt: None,
943            worker_id: None,
944            lease_expires_at_ms: None,
945        });
946}
947
948fn handoff_ready(ctx: &Ctx, state: &mut HandoffState, ready: &ScheduledReadinessReady) {
949    state
950        .ready_by_id
951        .insert(ready.schedule_id.clone(), ready.clone());
952    handoff_ready_inner(ctx, state, ready);
953}
954
955fn handoff_ready_inner(ctx: &Ctx, state: &mut HandoffState, ready: &ScheduledReadinessReady) {
956    let Some(queue_id) = ready_metadata_string(ready, "queueId") else {
957        emit_handoff_issue(
958            ctx,
959            state,
960            ready.schedule_id.clone(),
961            "missing queueId metadata",
962        );
963        return;
964    };
965    let Some(work_id) = ready_metadata_string(ready, "workId") else {
966        emit_handoff_issue(
967            ctx,
968            state,
969            ready.schedule_id.clone(),
970            "missing workId metadata",
971        );
972        return;
973    };
974    let Some(schedule_kind) = ready_metadata_string(ready, "scheduleKind")
975        .and_then(|value| WorkQueueReadinessScheduleKind::from_str(&value))
976    else {
977        emit_handoff_issue(
978            ctx,
979            state,
980            ready.schedule_id.clone(),
981            "missing scheduleKind metadata",
982        );
983        return;
984    };
985    let Some(work) = state
986        .works_by_id
987        .get(&work_key(&queue_id, &work_id))
988        .cloned()
989    else {
990        emit_handoff_status(
991            ctx,
992            state,
993            ignored_status(&queue_id, &work_id, ready, schedule_kind, "unknown-work"),
994        );
995        return;
996    };
997    if work.queue_id != queue_id {
998        emit_handoff_status(
999            ctx,
1000            state,
1001            ignored_status(&queue_id, &work_id, ready, schedule_kind, "queue-mismatch"),
1002        );
1003        return;
1004    }
1005    let candidate = match schedule_kind {
1006        WorkQueueReadinessScheduleKind::AdmissionDelay
1007        | WorkQueueReadinessScheduleKind::WorkScheduled
1008        | WorkQueueReadinessScheduleKind::RetryScheduled => {
1009            if work.effective_ready_at_ms != Some(ready.ready_at_ms) {
1010                emit_handoff_status(
1011                    ctx,
1012                    state,
1013                    ignored_status(
1014                        &queue_id,
1015                        &work_id,
1016                        ready,
1017                        schedule_kind,
1018                        "superseded-readiness",
1019                    ),
1020                );
1021                return;
1022            }
1023            if work
1024                .effective_schedule_kind
1025                .is_some_and(|effective| effective != schedule_kind)
1026                && matches!(
1027                    work.state,
1028                    WorkQueueDerivedState::Scheduled | WorkQueueDerivedState::RetryWait
1029                )
1030            {
1031                emit_handoff_status(
1032                    ctx,
1033                    state,
1034                    ignored_status(
1035                        &queue_id,
1036                        &work_id,
1037                        ready,
1038                        schedule_kind,
1039                        "stale-readiness-kind",
1040                    ),
1041                );
1042                return;
1043            }
1044            if matches!(
1045                work.state,
1046                WorkQueueDerivedState::Completed
1047                    | WorkQueueDerivedState::Canceled
1048                    | WorkQueueDerivedState::DeadLettered
1049                    | WorkQueueDerivedState::Leased
1050            ) {
1051                prune_candidates_for_work(state, &queue_id, &work_id);
1052                emit_handoff_status(
1053                    ctx,
1054                    state,
1055                    ignored_status(&queue_id, &work_id, ready, schedule_kind, "stale-readiness"),
1056                );
1057                return;
1058            }
1059            WorkQueueReadinessCandidate {
1060                candidate_id: compound_tuple_key(
1061                    "work-queue-readiness-candidate",
1062                    &[
1063                        &ready.schedule_id,
1064                        WorkQueueReadinessCandidateKind::ClaimEligible.as_str(),
1065                    ],
1066                ),
1067                queue_id,
1068                work_id,
1069                schedule_id: ready.schedule_id.clone(),
1070                candidate_kind: WorkQueueReadinessCandidateKind::ClaimEligible,
1071                schedule_kind,
1072                ready_at_ms: ready.ready_at_ms,
1073                now_ms: ready.now_ms,
1074                lease_id: None,
1075                attempt: None,
1076                worker_id: None,
1077                source_refs: ready.source_refs.clone(),
1078            }
1079        }
1080        WorkQueueReadinessScheduleKind::LeaseExpiration => {
1081            let lease_id = ready_metadata_string(ready, "leaseId");
1082            let attempt =
1083                ready_metadata_u64(ready, "attempt").and_then(|value| u32::try_from(value).ok());
1084            if work.state != WorkQueueDerivedState::Leased
1085                || work.lease_id != lease_id
1086                || work.attempt != attempt
1087                || work.lease_expires_at_ms != Some(ready.ready_at_ms)
1088            {
1089                prune_candidates_for_work(state, &queue_id, &work_id);
1090                emit_handoff_status(
1091                    ctx,
1092                    state,
1093                    ignored_status(
1094                        &queue_id,
1095                        &work_id,
1096                        ready,
1097                        schedule_kind,
1098                        "stale-lease-readiness",
1099                    ),
1100                );
1101                return;
1102            }
1103            WorkQueueReadinessCandidate {
1104                candidate_id: compound_tuple_key(
1105                    "work-queue-readiness-candidate",
1106                    &[
1107                        &ready.schedule_id,
1108                        WorkQueueReadinessCandidateKind::LeaseExpirationEligible.as_str(),
1109                    ],
1110                ),
1111                queue_id,
1112                work_id,
1113                schedule_id: ready.schedule_id.clone(),
1114                candidate_kind: WorkQueueReadinessCandidateKind::LeaseExpirationEligible,
1115                schedule_kind,
1116                ready_at_ms: ready.ready_at_ms,
1117                now_ms: ready.now_ms,
1118                lease_id,
1119                attempt,
1120                worker_id: work.worker_id,
1121                source_refs: ready.source_refs.clone(),
1122            }
1123        }
1124    };
1125    emit_candidate(ctx, state, candidate);
1126}
1127
1128fn work_key(queue_id: &str, work_id: &str) -> (String, String) {
1129    (queue_id.to_owned(), work_id.to_owned())
1130}
1131
1132fn prune_candidates_for_work(state: &mut HandoffState, queue_id: &str, work_id: &str) {
1133    state
1134        .candidates_by_id
1135        .retain(|_, candidate| candidate.queue_id != queue_id || candidate.work_id != work_id);
1136}
1137
1138fn handoff_overdue(ctx: &Ctx, state: &mut HandoffState, overdue: &ScheduledReadinessOverdue) {
1139    let queue_id = overdue_metadata_string(overdue, "queueId").unwrap_or_default();
1140    let work_id = overdue_metadata_string(overdue, "workId");
1141    let schedule_kind = overdue_metadata_string(overdue, "scheduleKind")
1142        .and_then(|value| WorkQueueReadinessScheduleKind::from_str(&value));
1143    emit_handoff_status(
1144        ctx,
1145        state,
1146        WorkQueueReadinessStatus {
1147            status_id: compound_tuple_key("work-queue-readiness-overdue", &[&overdue.schedule_id]),
1148            queue_id,
1149            work_id,
1150            schedule_id: Some(overdue.schedule_id.clone()),
1151            state: WorkQueueReadinessStatusState::Overdue,
1152            schedule_kind,
1153            ready_at_ms: Some(overdue.ready_at_ms),
1154            now_ms: Some(overdue.now_ms),
1155            issue_code: None,
1156            details: Some("deadline visibility only; no queue lifecycle mutation".to_owned()),
1157        },
1158    );
1159}
1160
1161fn emit_candidate(ctx: &Ctx, state: &mut HandoffState, candidate: WorkQueueReadinessCandidate) {
1162    state
1163        .candidates_by_id
1164        .insert(candidate.candidate_id.clone(), candidate.clone());
1165    if state
1166        .emitted
1167        .insert(compound_tuple_key("candidate", &[&candidate.candidate_id]))
1168    {
1169        emit_fact(ctx, WorkQueueReadinessFact::Candidate(candidate.clone()));
1170        emit_handoff_status(
1171            ctx,
1172            state,
1173            WorkQueueReadinessStatus {
1174                status_id: compound_tuple_key(
1175                    "work-queue-readiness-candidate-status",
1176                    &[&candidate.candidate_id],
1177                ),
1178                queue_id: candidate.queue_id.clone(),
1179                work_id: Some(candidate.work_id.clone()),
1180                schedule_id: Some(candidate.schedule_id.clone()),
1181                state: WorkQueueReadinessStatusState::Candidate,
1182                schedule_kind: Some(candidate.schedule_kind),
1183                ready_at_ms: Some(candidate.ready_at_ms),
1184                now_ms: Some(candidate.now_ms),
1185                issue_code: None,
1186                details: Some(format!("{:?}", candidate.candidate_kind)),
1187            },
1188        );
1189        emit_handoff_audit(ctx, state, "work-queue-readiness-candidate", &candidate);
1190    }
1191}
1192
1193fn emit_status(ctx: &Ctx, state: &mut TranslatorState, status: WorkQueueReadinessStatus) {
1194    state
1195        .status_by_id
1196        .insert(status.status_id.clone(), status.clone());
1197    if state
1198        .emitted
1199        .insert(compound_tuple_key("status", &[&format!("{status:?}")]))
1200    {
1201        emit_fact(ctx, WorkQueueReadinessFact::Status(status));
1202    }
1203}
1204
1205fn emit_handoff_status(ctx: &Ctx, state: &mut HandoffState, status: WorkQueueReadinessStatus) {
1206    state
1207        .status_by_id
1208        .insert(status.status_id.clone(), status.clone());
1209    if state
1210        .emitted
1211        .insert(compound_tuple_key("status", &[&format!("{status:?}")]))
1212    {
1213        emit_fact(ctx, WorkQueueReadinessFact::Status(status));
1214    }
1215}
1216
1217fn emit_handoff_issue(ctx: &Ctx, state: &mut HandoffState, schedule_id: String, detail: &str) {
1218    let issue = readiness_issue(
1219        "work-queue-readiness-malformed-ready",
1220        "workQueue readiness handoff requires workQueue metadata on ready facts.",
1221        &schedule_id,
1222        Some(detail.to_owned()),
1223        "error",
1224    );
1225    if state
1226        .emitted
1227        .insert(canonical_tuple_key(&["issue", &schedule_id, detail]))
1228    {
1229        emit_fact(ctx, WorkQueueReadinessFact::Issue(issue));
1230    }
1231}
1232
1233fn emit_translator_audit(
1234    ctx: &Ctx,
1235    state: &mut TranslatorState,
1236    kind: &str,
1237    schedule: &ScheduledReadinessRequested,
1238) {
1239    state.audit_seq += 1;
1240    emit_fact(
1241        ctx,
1242        WorkQueueReadinessFact::Audit(ScheduledReadinessAuditRecord {
1243            id: format!("work-queue-readiness-audit-{}", state.audit_seq),
1244            kind: kind.to_owned(),
1245            subject_id: Some(schedule.schedule_id.clone()),
1246            source_refs: schedule.source_refs.clone(),
1247            metadata: schedule.metadata.clone(),
1248        }),
1249    );
1250}
1251
1252fn emit_handoff_audit(
1253    ctx: &Ctx,
1254    state: &mut HandoffState,
1255    kind: &str,
1256    candidate: &WorkQueueReadinessCandidate,
1257) {
1258    state.audit_seq += 1;
1259    emit_fact(
1260        ctx,
1261        WorkQueueReadinessFact::Audit(ScheduledReadinessAuditRecord {
1262            id: format!("work-queue-readiness-handoff-audit-{}", state.audit_seq),
1263            kind: kind.to_owned(),
1264            subject_id: Some(candidate.schedule_id.clone()),
1265            source_refs: candidate.source_refs.clone(),
1266            metadata: Some(BTreeMap::from([
1267                (
1268                    "queueId".to_owned(),
1269                    JsonValue::from(candidate.queue_id.clone()),
1270                ),
1271                (
1272                    "workId".to_owned(),
1273                    JsonValue::from(candidate.work_id.clone()),
1274                ),
1275            ])),
1276        }),
1277    );
1278}
1279
1280fn ignored_status(
1281    queue_id: &str,
1282    work_id: &str,
1283    ready: &ScheduledReadinessReady,
1284    schedule_kind: WorkQueueReadinessScheduleKind,
1285    detail: &str,
1286) -> WorkQueueReadinessStatus {
1287    WorkQueueReadinessStatus {
1288        status_id: compound_tuple_key(
1289            "work-queue-readiness-ignored",
1290            &[&ready.schedule_id, detail],
1291        ),
1292        queue_id: queue_id.to_owned(),
1293        work_id: Some(work_id.to_owned()),
1294        schedule_id: Some(ready.schedule_id.clone()),
1295        state: WorkQueueReadinessStatusState::Ignored,
1296        schedule_kind: Some(schedule_kind),
1297        ready_at_ms: Some(ready.ready_at_ms),
1298        now_ms: Some(ready.now_ms),
1299        issue_code: None,
1300        details: Some(detail.to_owned()),
1301    }
1302}
1303
1304fn metadata_string(schedule: &ScheduledReadinessRequested, key: &str) -> Option<String> {
1305    schedule
1306        .metadata
1307        .as_ref()?
1308        .get(key)?
1309        .as_str()
1310        .map(str::to_owned)
1311}
1312
1313fn metadata_kind(schedule: &ScheduledReadinessRequested) -> Option<WorkQueueReadinessScheduleKind> {
1314    metadata_string(schedule, "scheduleKind")
1315        .as_deref()
1316        .and_then(WorkQueueReadinessScheduleKind::from_str)
1317}
1318
1319fn ready_metadata_string(ready: &ScheduledReadinessReady, key: &str) -> Option<String> {
1320    ready
1321        .metadata
1322        .as_ref()?
1323        .get(key)?
1324        .as_str()
1325        .map(str::to_owned)
1326}
1327
1328fn ready_metadata_u64(ready: &ScheduledReadinessReady, key: &str) -> Option<u64> {
1329    ready.metadata.as_ref()?.get(key)?.as_u64()
1330}
1331
1332fn overdue_metadata_string(overdue: &ScheduledReadinessOverdue, key: &str) -> Option<String> {
1333    overdue
1334        .metadata
1335        .as_ref()?
1336        .get(key)?
1337        .as_str()
1338        .map(str::to_owned)
1339}
1340
1341fn project_fact<T: Clone + 'static>(
1342    graph: &Graph,
1343    runtime: &Node<WorkQueueReadinessFact>,
1344    name: String,
1345    factory: &'static str,
1346    select: impl Fn(&WorkQueueReadinessFact) -> Option<T> + 'static,
1347) -> Node<T> {
1348    graph.node_opts::<T, _>(
1349        vec![runtime.erased()],
1350        move |ctx| {
1351            for fact in ctx.batch::<WorkQueueReadinessFact>(0) {
1352                if let Some(value) = select(&fact) {
1353                    ctx.emit(value);
1354                }
1355            }
1356        },
1357        node_opts(name, factory),
1358    )
1359}
1360
1361fn emit_fact(ctx: &Ctx, fact: WorkQueueReadinessFact) {
1362    ctx.emit(fact);
1363}
1364
1365fn node_opts(name: impl Into<String>, factory: impl Into<String>) -> GraphNodeOpts {
1366    let mut opts = GraphNodeOpts::named(name);
1367    opts.node = NodeOpts {
1368        factory: Some(factory.into()),
1369        complete_when_deps_complete: false,
1370        error_when_deps_error: false,
1371        ..opts.node
1372    };
1373    opts
1374}
graphrefly/work_queue/readiness.rs

graphrefly/work_queue/
readiness.rs
