graphrefly/cqrs/
work_queue.rs

1//! Optional CQRS-over-workQueue recipe (D350/D352/D353).
2//!
3//! Queue lifecycle remains worker disposition. CQRS truth remains the ordered
4//! event/status/error facts emitted by the CQRS core.
5
6use std::collections::{BTreeMap, BTreeSet};
7
8use crate::cqrs::{CqrsCommand, CqrsError, CqrsErrorCode, CqrsStatus, CqrsStatusState};
9use crate::ctx::Ctx;
10use crate::graph::{Graph, GraphNodeOpts};
11use crate::messaging::DataIssue;
12use crate::node::Node;
13use crate::work_queue::{WorkQueueCommand, WorkQueueRecord};
14
15#[derive(Debug, Clone, PartialEq)]
16/// `CqrsQueuedCommandPayload` data container.
17pub struct CqrsQueuedCommandPayload<TCommand> {
18    /// `kind` field for kind.
19    pub kind: String,
20    /// `command` field for command.
21    pub command: CqrsCommand<TCommand>,
22    /// `idempotency_key` field for idempotency key.
23    pub idempotency_key: Option<String>,
24    /// `source_refs` field for source refs.
25    pub source_refs: Vec<String>,
26    /// `policy_refs` field for policy refs.
27    pub policy_refs: Vec<String>,
28    /// `actor_refs` field for actor refs.
29    pub actor_refs: Vec<String>,
30    /// `audit_refs` field for audit refs.
31    pub audit_refs: Vec<String>,
32    /// `metadata` field for metadata.
33    pub metadata: Option<String>,
34}
35
36impl<TCommand> CqrsQueuedCommandPayload<TCommand> {
37    /// Creates or computes `new`.
38    pub fn new(command: CqrsCommand<TCommand>) -> Self {
39        Self {
40            kind: "cqrs-queued-command".to_owned(),
41            command,
42            idempotency_key: None,
43            source_refs: Vec::new(),
44            policy_refs: Vec::new(),
45            actor_refs: Vec::new(),
46            audit_refs: Vec::new(),
47            metadata: None,
48        }
49    }
50}
51
52#[derive(Debug, Clone, PartialEq)]
53/// `CqrsWorkQueueAttempt` data container.
54pub struct CqrsWorkQueueAttempt<TCommand> {
55    /// `kind` field for kind.
56    pub kind: String,
57    /// `work_id` field for work id.
58    pub work_id: String,
59    /// `lease_id` field for lease id.
60    pub lease_id: String,
61    /// `queue_attempt` field for queue attempt.
62    pub queue_attempt: u32,
63    /// `worker_id` field for worker id.
64    pub worker_id: String,
65    /// `command` field for command.
66    pub command: CqrsCommand<TCommand>,
67    /// `payload` field for payload.
68    pub payload: CqrsQueuedCommandPayload<TCommand>,
69    /// `source_refs` field for source refs.
70    pub source_refs: Vec<String>,
71}
72
73#[derive(Debug, Clone, PartialEq)]
74/// `CqrsWorkQueueOutcome` variants.
75pub enum CqrsWorkQueueOutcome<TCommand> {
76    /// `Accepted` variant.
77    Accepted {
78        /// `status` field for status.
79        status: CqrsStatus,
80    },
81    /// `Rejected` variant.
82    Rejected {
83        /// `status` field for status.
84        status: CqrsStatus,
85        /// `error` field for error.
86        error: Option<Box<CqrsError<TCommand>>>,
87    },
88    /// `Release` variant.
89    Release {
90        /// `reason` field for reason.
91        reason: Option<String>,
92    },
93}
94
95#[derive(Clone, Default)]
96/// `CqrsWorkQueuePolicy` data container.
97pub struct CqrsWorkQueuePolicy {
98    /// `deterministic_handler_failures` field for deterministic handler failures.
99    pub deterministic_handler_failures: Vec<CqrsErrorCode>,
100}
101
102impl CqrsWorkQueuePolicy {
103    /// Updates or reads `deterministic_handler_failure`.
104    pub fn deterministic_handler_failure(mut self, code: CqrsErrorCode) -> Self {
105        if !self.deterministic_handler_failures.contains(&code) {
106            self.deterministic_handler_failures.push(code);
107        }
108        self
109    }
110
111    fn retryable_failure(&self, code: Option<CqrsErrorCode>) -> bool {
112        match code {
113            Some(code @ (CqrsErrorCode::HandlerThrew | CqrsErrorCode::ClockThrew)) => {
114                !self.deterministic_handler_failures.contains(&code)
115            }
116            None => true,
117            Some(_) => false,
118        }
119    }
120}
121
122#[derive(Clone)]
123/// `CqrsWorkQueueRecipeOptions` data container.
124pub struct CqrsWorkQueueRecipeOptions<TCommand> {
125    /// `name` field for name.
126    pub name: String,
127    /// `records` field for records.
128    pub records: Node<WorkQueueRecord<CqrsQueuedCommandPayload<TCommand>>>,
129    /// `status` field for status.
130    pub status: Node<CqrsStatus>,
131    /// `errors` field for errors.
132    pub errors: Option<Node<CqrsError<TCommand>>>,
133    /// `worker_id` field for worker id.
134    pub worker_id: Option<String>,
135    /// `policy` field for policy.
136    pub policy: CqrsWorkQueuePolicy,
137}
138
139impl<TCommand> CqrsWorkQueueRecipeOptions<TCommand> {
140    /// Creates or computes `new`.
141    pub fn new(
142        records: Node<WorkQueueRecord<CqrsQueuedCommandPayload<TCommand>>>,
143        status: Node<CqrsStatus>,
144    ) -> Self {
145        Self {
146            name: "cqrsWorkQueue".to_owned(),
147            records,
148            status,
149            errors: None,
150            worker_id: None,
151            policy: CqrsWorkQueuePolicy::default(),
152        }
153    }
154
155    /// Updates or reads `named`.
156    pub fn named(mut self, name: impl Into<String>) -> Self {
157        self.name = name.into();
158        self
159    }
160
161    /// Updates or reads `with_errors`.
162    pub fn with_errors(mut self, errors: Node<CqrsError<TCommand>>) -> Self {
163        self.errors = Some(errors);
164        self
165    }
166
167    /// Updates or reads `for_worker`.
168    pub fn for_worker(mut self, worker_id: impl Into<String>) -> Self {
169        self.worker_id = Some(worker_id.into());
170        self
171    }
172
173    /// Updates or reads `with_policy`.
174    pub fn with_policy(mut self, policy: CqrsWorkQueuePolicy) -> Self {
175        self.policy = policy;
176        self
177    }
178}
179
180#[derive(Clone)]
181/// `CqrsWorkQueueRecipeBundle` data container.
182pub struct CqrsWorkQueueRecipeBundle<TCommand> {
183    /// `attempts` field for attempts.
184    pub attempts: Node<CqrsWorkQueueAttempt<TCommand>>,
185    /// `dispatches` field for dispatches.
186    pub dispatches: Node<CqrsCommand<TCommand>>,
187    /// `commands` field for commands.
188    pub commands: Node<WorkQueueCommand<CqrsQueuedCommandPayload<TCommand>>>,
189    /// `issues` field for issues.
190    pub issues: Node<DataIssue>,
191}
192
193#[derive(Clone)]
194enum CqrsQueueFact<TCommand> {
195    Attempt(CqrsWorkQueueAttempt<TCommand>),
196    Dispatch(CqrsCommand<TCommand>),
197    Command(WorkQueueCommand<CqrsQueuedCommandPayload<TCommand>>),
198    Issue(DataIssue),
199}
200
201#[derive(Clone)]
202struct CqrsQueueState<TCommand> {
203    payloads: BTreeMap<String, CqrsQueuedCommandPayload<TCommand>>,
204    active_claims: BTreeMap<String, Vec<CqrsWorkQueueAttempt<TCommand>>>,
205    errors: BTreeMap<String, CqrsError<TCommand>>,
206    terminal_claims: BTreeSet<String>,
207}
208
209impl<TCommand> Default for CqrsQueueState<TCommand> {
210    fn default() -> Self {
211        Self {
212            payloads: BTreeMap::new(),
213            active_claims: BTreeMap::new(),
214            errors: BTreeMap::new(),
215            terminal_claims: BTreeSet::new(),
216        }
217    }
218}
219
220/// Creates or computes `cqrs_work_queue_recipe`.
221pub fn cqrs_work_queue_recipe<TCommand: Clone + 'static>(
222    graph: &Graph,
223    opts: CqrsWorkQueueRecipeOptions<TCommand>,
224) -> CqrsWorkQueueRecipeBundle<TCommand> {
225    let name = opts.name.clone();
226    let mut deps = vec![opts.records.erased(), opts.status.erased()];
227    if let Some(errors) = &opts.errors {
228        deps.push(errors.erased());
229    }
230    let runtime = graph.node_opts::<CqrsQueueFact<TCommand>, _>(
231        deps,
232        move |ctx| {
233            let mut state = ctx
234                .state_get::<CqrsQueueState<TCommand>>()
235                .map(|state| (*state).clone())
236                .unwrap_or_default();
237            for record in ctx.batch::<WorkQueueRecord<CqrsQueuedCommandPayload<TCommand>>>(0) {
238                reduce_record(ctx, &mut state, &record, &opts);
239            }
240            if opts.errors.is_some() {
241                for error in ctx.batch::<CqrsError<TCommand>>(2) {
242                    if let Some(command) = &error.command {
243                        state.errors.insert(command.id.clone(), (*error).clone());
244                    }
245                }
246            }
247            for status in ctx.batch::<CqrsStatus>(1) {
248                reduce_status(ctx, &mut state, &status, &opts);
249            }
250            ctx.state_set(state);
251            ctx.state_persist(true);
252        },
253        GraphNodeOpts::named(format!("{name}/runtime")),
254    );
255    CqrsWorkQueueRecipeBundle {
256        attempts: project(
257            graph,
258            &runtime,
259            format!("{name}/attempts"),
260            |fact| match fact {
261                CqrsQueueFact::Attempt(attempt) => Some(attempt.clone()),
262                _ => None,
263            },
264        ),
265        dispatches: project(
266            graph,
267            &runtime,
268            format!("{name}/dispatches"),
269            |fact| match fact {
270                CqrsQueueFact::Dispatch(command) => Some(command.clone()),
271                _ => None,
272            },
273        ),
274        commands: project(
275            graph,
276            &runtime,
277            format!("{name}/commands"),
278            |fact| match fact {
279                CqrsQueueFact::Command(command) => Some(command.clone()),
280                _ => None,
281            },
282        ),
283        issues: project(
284            graph,
285            &runtime,
286            format!("{name}/issues"),
287            |fact| match fact {
288                CqrsQueueFact::Issue(issue) => Some(issue.clone()),
289                _ => None,
290            },
291        ),
292    }
293}
294
295/// Creates or computes `cqrs_submit_command`.
296pub fn cqrs_submit_command<TCommand: Clone>(
297    command: CqrsCommand<TCommand>,
298) -> WorkQueueCommand<CqrsQueuedCommandPayload<TCommand>> {
299    let command_id = format!("{}:cqrs-work-queue-submit", command.id);
300    let mut payload = CqrsQueuedCommandPayload::new(command.clone());
301    payload.idempotency_key = Some(command.id.clone());
302    WorkQueueCommand::Submit {
303        payload,
304        command_id,
305        queue_id: None,
306        idempotency_key: Some(command.id),
307    }
308}
309
310/// Creates or computes `cqrs_work_queue_disposition_command`.
311pub fn cqrs_work_queue_disposition_command<TCommand: Clone>(
312    attempt: &CqrsWorkQueueAttempt<TCommand>,
313    outcome: CqrsWorkQueueOutcome<TCommand>,
314    policy: &CqrsWorkQueuePolicy,
315) -> WorkQueueCommand<CqrsQueuedCommandPayload<TCommand>> {
316    match outcome {
317        CqrsWorkQueueOutcome::Release { reason } => WorkQueueCommand::Release {
318            command_id: disposition_command_id(attempt, "release"),
319            queue_id: None,
320            idempotency_key: None,
321            work_id: attempt.work_id.clone(),
322            lease_id: attempt.lease_id.clone(),
323            attempt: attempt.queue_attempt,
324            worker_id: attempt.worker_id.clone(),
325            reason,
326            now_ms: None,
327        },
328        CqrsWorkQueueOutcome::Accepted { status } => WorkQueueCommand::Complete {
329            command_id: disposition_command_id(attempt, "complete"),
330            queue_id: None,
331            idempotency_key: None,
332            work_id: attempt.work_id.clone(),
333            lease_id: attempt.lease_id.clone(),
334            attempt: attempt.queue_attempt,
335            worker_id: attempt.worker_id.clone(),
336            result: Some(format!(
337                "cqrs-accepted:command_id={};event_count={}",
338                status.command_id.unwrap_or_default(),
339                status.event_count
340            )),
341            now_ms: None,
342        },
343        CqrsWorkQueueOutcome::Rejected { status, error } => {
344            let code = status.error_code;
345            if matches!(
346                code,
347                Some(CqrsErrorCode::HandlerThrew | CqrsErrorCode::ClockThrew) | None
348            ) {
349                WorkQueueCommand::Fail {
350                    command_id: disposition_command_id(attempt, "fail"),
351                    queue_id: None,
352                    idempotency_key: None,
353                    work_id: attempt.work_id.clone(),
354                    lease_id: attempt.lease_id.clone(),
355                    attempt: attempt.queue_attempt,
356                    worker_id: attempt.worker_id.clone(),
357                    error: Some(error_message(code, error.as_deref())),
358                    retryable: Some(policy.retryable_failure(code)),
359                    now_ms: None,
360                }
361            } else {
362                WorkQueueCommand::Complete {
363                    command_id: disposition_command_id(attempt, "complete"),
364                    queue_id: None,
365                    idempotency_key: None,
366                    work_id: attempt.work_id.clone(),
367                    lease_id: attempt.lease_id.clone(),
368                    attempt: attempt.queue_attempt,
369                    worker_id: attempt.worker_id.clone(),
370                    result: Some(format!(
371                        "cqrs-rejected:command_id={};error_code={:?};event_count={}",
372                        status.command_id.unwrap_or_default(),
373                        code,
374                        status.event_count
375                    )),
376                    now_ms: None,
377                }
378            }
379        }
380    }
381}
382
383fn reduce_record<TCommand: Clone + 'static>(
384    ctx: &Ctx,
385    state: &mut CqrsQueueState<TCommand>,
386    record: &WorkQueueRecord<CqrsQueuedCommandPayload<TCommand>>,
387    opts: &CqrsWorkQueueRecipeOptions<TCommand>,
388) {
389    match record {
390        WorkQueueRecord::WorkAdmitted {
391            work_id, payload, ..
392        } => {
393            if payload.kind == "cqrs-queued-command" {
394                state.payloads.insert(work_id.clone(), payload.clone());
395            } else {
396                ctx.emit(CqrsQueueFact::<TCommand>::Issue(queue_issue(
397                    record,
398                    "cqrs-queue-malformed-payload",
399                )));
400            }
401        }
402        WorkQueueRecord::WorkClaimed {
403            work_id,
404            lease_id,
405            attempt,
406            worker_id,
407            record_seq,
408            ..
409        } => {
410            if opts
411                .worker_id
412                .as_deref()
413                .is_some_and(|expected| expected != worker_id)
414            {
415                return;
416            }
417            let Some(payload) = state.payloads.get(work_id).cloned() else {
418                ctx.emit(CqrsQueueFact::<TCommand>::Issue(queue_issue(
419                    record,
420                    "cqrs-claim-without-payload",
421                )));
422                ctx.emit(CqrsQueueFact::<TCommand>::Command(
423                    WorkQueueCommand::Release {
424                        command_id: format!(
425                            "cqrs:{work_id}:{lease_id}:{attempt}:release-no-payload"
426                        ),
427                        queue_id: None,
428                        idempotency_key: None,
429                        work_id: work_id.clone(),
430                        lease_id: lease_id.clone(),
431                        attempt: *attempt,
432                        worker_id: worker_id.clone(),
433                        reason: Some("cqrs-claim-without-payload".to_owned()),
434                        now_ms: None,
435                    },
436                ));
437                return;
438            };
439            let attempt_fact = CqrsWorkQueueAttempt {
440                kind: "cqrs-work-queue-attempt".to_owned(),
441                work_id: work_id.clone(),
442                lease_id: lease_id.clone(),
443                queue_attempt: *attempt,
444                worker_id: worker_id.clone(),
445                command: payload.command.clone(),
446                payload,
447                source_refs: vec![format!("work-queue-record:{record_seq}")],
448            };
449            state
450                .active_claims
451                .entry(attempt_fact.command.id.clone())
452                .or_default()
453                .push(attempt_fact.clone());
454            ctx.emit(CqrsQueueFact::<TCommand>::Attempt(attempt_fact.clone()));
455            ctx.emit(CqrsQueueFact::<TCommand>::Dispatch(attempt_fact.command));
456        }
457        _ => invalidate_claims_for_record(state, record),
458    }
459}
460
461fn invalidate_claims_for_record<TCommand: Clone>(
462    state: &mut CqrsQueueState<TCommand>,
463    record: &WorkQueueRecord<CqrsQueuedCommandPayload<TCommand>>,
464) {
465    match record {
466        WorkQueueRecord::WorkReleased {
467            work_id,
468            lease_id,
469            attempt,
470            worker_id,
471            ..
472        }
473        | WorkQueueRecord::LeaseExpired {
474            work_id,
475            lease_id,
476            attempt,
477            worker_id,
478            ..
479        }
480        | WorkQueueRecord::AttemptFailed {
481            work_id,
482            lease_id,
483            attempt,
484            worker_id,
485            ..
486        }
487        | WorkQueueRecord::AttemptCompleted {
488            work_id,
489            lease_id,
490            attempt,
491            worker_id,
492            ..
493        }
494        | WorkQueueRecord::WorkCompleted {
495            work_id,
496            lease_id,
497            attempt,
498            worker_id,
499            ..
500        } => retain_active_claims(state, |claim| {
501            claim.work_id != *work_id
502                || claim.lease_id != *lease_id
503                || claim.queue_attempt != *attempt
504                || claim.worker_id != *worker_id
505        }),
506        WorkQueueRecord::WorkDeadLettered { work_id, .. } => {
507            retain_active_claims(state, |claim| claim.work_id != *work_id);
508        }
509        WorkQueueRecord::WorkCanceled {
510            work_id,
511            canceled_lease_id,
512            attempt,
513            ..
514        } => retain_active_claims(state, |claim| {
515            if claim.work_id != *work_id {
516                return true;
517            }
518            if let Some(canceled_lease_id) = canceled_lease_id {
519                if claim.lease_id != *canceled_lease_id {
520                    return true;
521                }
522            }
523            if let Some(attempt) = attempt {
524                if claim.queue_attempt != *attempt {
525                    return true;
526                }
527            }
528            false
529        }),
530        WorkQueueRecord::RetryScheduled { .. }
531        | WorkQueueRecord::WorkAdmitted { .. }
532        | WorkQueueRecord::AdmissionDeduped { .. }
533        | WorkQueueRecord::WorkScheduled { .. }
534        | WorkQueueRecord::WorkClaimed { .. }
535        | WorkQueueRecord::LeaseRenewed { .. } => {}
536    }
537}
538
539fn retain_active_claims<TCommand: Clone>(
540    state: &mut CqrsQueueState<TCommand>,
541    keep: impl Fn(&CqrsWorkQueueAttempt<TCommand>) -> bool,
542) {
543    state.active_claims.retain(|_, claims| {
544        claims.retain(|claim| keep(claim));
545        !claims.is_empty()
546    });
547}
548
549fn reduce_status<TCommand: Clone + 'static>(
550    ctx: &Ctx,
551    state: &mut CqrsQueueState<TCommand>,
552    status: &CqrsStatus,
553    opts: &CqrsWorkQueueRecipeOptions<TCommand>,
554) {
555    let Some(command_id) = &status.command_id else {
556        return;
557    };
558    let Some(attempt) = shift_active_claim(&mut state.active_claims, command_id) else {
559        ctx.emit(CqrsQueueFact::<TCommand>::Issue(DataIssue {
560            kind: "issue".to_owned(),
561            code: "cqrs-status-without-active-queue-claim".to_owned(),
562            message: "CQRS workQueue recipe observed status without an active queue claim"
563                .to_owned(),
564            severity: "error".to_owned(),
565            source: "cqrs.workQueue".to_owned(),
566            topic: None,
567            details: Some(format!("cqrs-command:{command_id}")),
568        }));
569        return;
570    };
571    let claim_key = format!(
572        "{}:{}:{}",
573        attempt.work_id, attempt.lease_id, attempt.queue_attempt
574    );
575    if !state.terminal_claims.insert(claim_key.clone()) {
576        ctx.emit(CqrsQueueFact::<TCommand>::Issue(DataIssue {
577            kind: "issue".to_owned(),
578            code: "cqrs-duplicate-terminal-outcome-for-queue-claim".to_owned(),
579            message: format!(
580                "CQRS queue claim '{claim_key}' already produced a terminal disposition"
581            ),
582            severity: "error".to_owned(),
583            source: "cqrs.workQueue".to_owned(),
584            topic: None,
585            details: Some(format!("cqrs-command:{command_id}")),
586        }));
587        return;
588    }
589    let outcome = if status.state == CqrsStatusState::Accepted {
590        CqrsWorkQueueOutcome::Accepted {
591            status: status.clone(),
592        }
593    } else {
594        CqrsWorkQueueOutcome::Rejected {
595            status: status.clone(),
596            error: state.errors.get(command_id).cloned().map(Box::new),
597        }
598    };
599    ctx.emit(CqrsQueueFact::<TCommand>::Command(
600        cqrs_work_queue_disposition_command(&attempt, outcome, &opts.policy),
601    ));
602}
603
604fn shift_active_claim<TCommand>(
605    claims: &mut BTreeMap<String, Vec<CqrsWorkQueueAttempt<TCommand>>>,
606    command_id: &str,
607) -> Option<CqrsWorkQueueAttempt<TCommand>> {
608    let queue = claims.get_mut(command_id)?;
609    if queue.is_empty() {
610        return None;
611    }
612    let first = queue.remove(0);
613    if queue.is_empty() {
614        claims.remove(command_id);
615    }
616    Some(first)
617}
618
619fn project<TIn: Clone + 'static, TOut: 'static>(
620    graph: &Graph,
621    source: &Node<TIn>,
622    name: String,
623    pick: impl Fn(&TIn) -> Option<TOut> + 'static,
624) -> Node<TOut> {
625    graph.node_opts::<TOut, _>(
626        vec![source.erased()],
627        move |ctx| {
628            for fact in ctx.batch::<TIn>(0) {
629                if let Some(value) = pick(&fact) {
630                    ctx.emit(value);
631                }
632            }
633        },
634        GraphNodeOpts::named(name),
635    )
636}
637
638fn disposition_command_id<TCommand>(
639    attempt: &CqrsWorkQueueAttempt<TCommand>,
640    suffix: &str,
641) -> String {
642    format!(
643        "cqrs:{}:{}:{}:{suffix}",
644        attempt.work_id, attempt.lease_id, attempt.queue_attempt
645    )
646}
647
648fn error_message<TCommand>(
649    code: Option<CqrsErrorCode>,
650    error: Option<&CqrsError<TCommand>>,
651) -> String {
652    error
653        .map(|error| error.message.clone())
654        .unwrap_or_else(|| format!("cqrs rejected with {code:?}"))
655}
656
657fn queue_issue<TCommand>(record: &WorkQueueRecord<TCommand>, code: impl Into<String>) -> DataIssue {
658    DataIssue {
659        kind: "issue".to_owned(),
660        code: code.into(),
661        message: format!(
662            "CQRS workQueue recipe could not map workQueue record '{}'",
663            record_kind(record)
664        ),
665        severity: "error".to_owned(),
666        source: "cqrs.workQueue".to_owned(),
667        topic: None,
668        details: Some(format!(
669            "work_id={};record_seq={}",
670            record.work_id(),
671            record.record_seq()
672        )),
673    }
674}
675
676fn record_kind<TCommand>(record: &WorkQueueRecord<TCommand>) -> &'static str {
677    match record {
678        WorkQueueRecord::WorkAdmitted { .. } => "work-admitted",
679        WorkQueueRecord::AdmissionDeduped { .. } => "admission-deduped",
680        WorkQueueRecord::WorkScheduled { .. } => "work-scheduled",
681        WorkQueueRecord::WorkClaimed { .. } => "work-claimed",
682        WorkQueueRecord::LeaseRenewed { .. } => "lease-renewed",
683        WorkQueueRecord::WorkReleased { .. } => "work-released",
684        WorkQueueRecord::LeaseExpired { .. } => "lease-expired",
685        WorkQueueRecord::AttemptCompleted { .. } => "attempt-completed",
686        WorkQueueRecord::WorkCompleted { .. } => "work-completed",
687        WorkQueueRecord::AttemptFailed { .. } => "attempt-failed",
688        WorkQueueRecord::RetryScheduled { .. } => "retry-scheduled",
689        WorkQueueRecord::WorkDeadLettered { .. } => "work-dead-lettered",
690        WorkQueueRecord::WorkCanceled { .. } => "work-canceled",
691    }
692}
graphrefly/cqrs/work_queue.rs

graphrefly/cqrs/
work_queue.rs
