cronoq_server/
convert.rs

1//! Conversions between protobuf messages and scheduler types, with input validation.
2//!
3//! Everything a client or worker sends is validated here before it reaches the scheduler:
4//! strings must not contain NUL bytes (PostgreSQL cannot store them) and must respect the
5//! configured length and count limits.
6
7use std::sync::Arc;
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use bytes::Bytes;
11use cronoq::{
12    Clock, Dep, EdgeCond, Error, MissPolicy, OverlapPolicy, Priority, RetryPolicy, Snapshot,
13    StatusEvent, SubmitError, Task, TaskId, TaskRef, TaskStatus, When,
14};
15use cronoq_proto::v1 as pb;
16use tonic::Status;
17
18use crate::config::LimitsConfig;
19
20/// Tag carrying the payload content type.
21pub const CONTENT_TYPE_TAG: &str = "cronoq.content_type";
22
23/// Longest duration accepted from clients (100 years).
24pub const MAX_SECS: f64 = 100.0 * 365.0 * 86400.0;
25
26/// Longest error message accepted from a worker (longer ones are truncated).
27pub const MAX_MESSAGE_LEN: usize = 8192;
28
29/// Longest content type.
30const MAX_CONTENT_TYPE_LEN: usize = 128;
31
32/// Validate a client-supplied duration in seconds.
33pub fn secs(s: f64) -> Result<Duration, Status> {
34    if !s.is_finite() || !(0.0..=MAX_SECS).contains(&s) {
35        return Err(Status::invalid_argument(format!("invalid duration {s}")));
36    }
37    Ok(Duration::from_secs_f64(s))
38}
39
40/// Validate a client-supplied Unix timestamp in nanoseconds.
41pub fn unix(field: &str, ns: i64) -> Result<SystemTime, Status> {
42    if ns < 0 {
43        return Err(Status::invalid_argument(format!(
44            "{field} must not be negative"
45        )));
46    }
47    Ok(UNIX_EPOCH + Duration::from_nanos(ns as u64))
48}
49
50/// A wall-clock time as Unix nanoseconds (saturating).
51pub fn unix_ns(t: SystemTime) -> i64 {
52    match t.duration_since(UNIX_EPOCH) {
53        Ok(d) => i64::try_from(d.as_nanos()).unwrap_or(i64::MAX),
54        Err(_) => 0,
55    }
56}
57
58/// Validate a client- or worker-supplied string.
59pub fn check_str(field: &str, s: &str, max_len: usize) -> Result<(), Status> {
60    if s.len() > max_len {
61        return Err(Status::invalid_argument(format!(
62            "{field} is longer than {max_len} bytes"
63        )));
64    }
65    if s.contains('\0') {
66        return Err(Status::invalid_argument(format!(
67            "{field} must not contain NUL characters"
68        )));
69    }
70    Ok(())
71}
72
73/// Make a worker-supplied message safe to store and log: NUL bytes are replaced and the text is
74/// truncated (on a character boundary) to [`MAX_MESSAGE_LEN`].
75pub fn sanitize_message(s: &str) -> String {
76    let mut out: String = s
77        .chars()
78        .map(|c| if c == '\0' { '\u{FFFD}' } else { c })
79        .collect();
80    if out.len() > MAX_MESSAGE_LEN {
81        let mut end = MAX_MESSAGE_LEN;
82        while !out.is_char_boundary(end) {
83            end -= 1;
84        }
85        out.truncate(end);
86    }
87    out
88}
89
90/// Make a worker-supplied name safe to log: control characters are dropped, length capped.
91pub fn sanitize_name(s: &str) -> String {
92    s.chars().filter(|c| !c.is_control()).take(64).collect()
93}
94
95/// Map a wire priority.
96pub fn priority(p: i32) -> Result<Priority, Status> {
97    Ok(
98        match pb::Priority::try_from(p).map_err(|_| Status::invalid_argument("unknown priority"))? {
99            pb::Priority::Critical => Priority::Critical,
100            pb::Priority::High => Priority::High,
101            pb::Priority::Unspecified | pb::Priority::Normal => Priority::Normal,
102            pb::Priority::Low => Priority::Low,
103        },
104    )
105}
106
107/// Map an engine priority to the wire.
108pub fn priority_to_pb(p: Priority) -> pb::Priority {
109    match p {
110        Priority::Critical => pb::Priority::Critical,
111        Priority::High => pb::Priority::High,
112        Priority::Normal => pb::Priority::Normal,
113        Priority::Low => pb::Priority::Low,
114    }
115}
116
117/// Map a wire task reference.
118pub fn task_ref(r: Option<pb::TaskRef>, limits: &LimitsConfig) -> Result<TaskRef, Status> {
119    match r.and_then(|r| r.target) {
120        Some(pb::task_ref::Target::Id(id)) => Ok(TaskRef::Id(TaskId(id))),
121        Some(pb::task_ref::Target::Key(k)) if !k.is_empty() => {
122            check_str("key", &k, limits.max_key_len)?;
123            Ok(TaskRef::Key(k.into()))
124        }
125        _ => Err(Status::invalid_argument(
126            "task reference requires an id or key",
127        )),
128    }
129}
130
131/// Map a wire retry policy (zero fields take the defaults).
132pub fn retry(r: &pb::RetryPolicy) -> Result<RetryPolicy, Status> {
133    let d = RetryPolicy::default();
134    if !r.multiplier.is_finite() || r.multiplier < 0.0 || r.multiplier > 1000.0 {
135        return Err(Status::invalid_argument(
136            "retry.multiplier must be between 1 and 1000",
137        ));
138    }
139    Ok(RetryPolicy {
140        max_attempts: r.max_attempts.max(1),
141        initial_backoff: if r.initial_backoff_s != 0.0 {
142            secs(r.initial_backoff_s)?
143        } else {
144            d.initial_backoff
145        },
146        multiplier: if r.multiplier > 0.0 {
147            r.multiplier.max(1.0)
148        } else {
149            d.multiplier
150        },
151        max_backoff: if r.max_backoff_s != 0.0 {
152            secs(r.max_backoff_s)?
153        } else {
154            d.max_backoff
155        },
156        jitter: r.jitter,
157    })
158}
159
160fn edge_cond(c: i32) -> Result<EdgeCond, Status> {
161    Ok(
162        match pb::EdgeCond::try_from(c)
163            .map_err(|_| Status::invalid_argument("unknown dependency condition"))?
164        {
165            pb::EdgeCond::Unspecified | pb::EdgeCond::OnSuccess => EdgeCond::OnSuccess,
166            pb::EdgeCond::OnComplete => EdgeCond::OnComplete,
167            pb::EdgeCond::OnFailure => EdgeCond::OnFailure,
168        },
169    )
170}
171
172fn miss_policy(m: i32) -> Result<MissPolicy, Status> {
173    Ok(
174        match pb::MissPolicy::try_from(m)
175            .map_err(|_| Status::invalid_argument("unknown miss policy"))?
176        {
177            pb::MissPolicy::Unspecified | pb::MissPolicy::RunLate => MissPolicy::RunLate,
178            pb::MissPolicy::Skip => MissPolicy::Skip,
179            pb::MissPolicy::Fail => MissPolicy::Fail,
180        },
181    )
182}
183
184/// The pool a wire spec targets (`default` when empty).
185pub fn spec_pool(spec: &pb::TaskSpec) -> &str {
186    if spec.pool.is_empty() {
187        "default"
188    } else {
189        &spec.pool
190    }
191}
192
193/// Whether `key` has the form of a recurring occurrence's key, `<prefix>@<UTC RFC 3339 time>`.
194pub fn is_occurrence_key(key: &str) -> bool {
195    let Some((_, ts)) = key.rsplit_once('@') else {
196        return false;
197    };
198    let b = ts.as_bytes();
199    b.len() == 20
200        && b.iter().enumerate().all(|(i, c)| match i {
201            4 | 7 => *c == b'-',
202            10 => *c == b'T',
203            13 | 16 => *c == b':',
204            19 => *c == b'Z',
205            _ => c.is_ascii_digit(),
206        })
207}
208
209/// Build a task from its wire form, enforcing `limits`.
210pub fn task(spec: pb::TaskSpec, limits: &LimitsConfig) -> Result<Task<Bytes>, Status> {
211    check_str("kind", &spec.kind, limits.max_kind_len)?;
212    check_str("pool", &spec.pool, 64)?;
213    check_str("key", &spec.key, limits.max_key_len)?;
214    if is_occurrence_key(&spec.key) {
215        // Reserved, so a client cannot take a future occurrence's key in advance.
216        return Err(Status::invalid_argument(
217            "keys ending in '@<UTC time>' are reserved for recurring occurrences",
218        ));
219    }
220    check_str("content_type", &spec.content_type, MAX_CONTENT_TYPE_LEN)?;
221    if spec.tags.len() > limits.max_tags {
222        return Err(Status::invalid_argument(format!(
223            "at most {} tags are allowed",
224            limits.max_tags
225        )));
226    }
227    if spec.labels.len() > limits.max_tags {
228        return Err(Status::invalid_argument(format!(
229            "at most {} labels are allowed",
230            limits.max_tags
231        )));
232    }
233    if spec.deps.len() > limits.max_deps {
234        return Err(Status::invalid_argument(format!(
235            "at most {} dependencies are allowed",
236            limits.max_deps
237        )));
238    }
239    if spec.resources.len() > limits.max_resources {
240        return Err(Status::invalid_argument(format!(
241            "at most {} resources are allowed",
242            limits.max_resources
243        )));
244    }
245    let mut t = Task::new(Bytes::from(spec.payload)).priority(priority(spec.priority)?);
246    if !spec.kind.is_empty() {
247        t = t.kind(spec.kind);
248    }
249    if !spec.pool.is_empty() {
250        t = t.pool(spec.pool);
251    }
252    if !spec.key.is_empty() {
253        t = t.key(spec.key);
254    }
255    match (spec.deadline_in_s, spec.deadline_unix_ns) {
256        (Some(_), Some(_)) => {
257            return Err(Status::invalid_argument(
258                "set deadline_in_s or deadline_unix_ns, not both",
259            ));
260        }
261        (Some(s), None) => t = t.deadline_in(secs(s)?),
262        (None, Some(ns)) => t = t.deadline_at(unix("deadline_unix_ns", ns)?),
263        (None, None) => {}
264    }
265    let start = match (spec.start_in_s, spec.start_unix_ns) {
266        (Some(_), Some(_)) => {
267            return Err(Status::invalid_argument(
268                "set start_in_s or start_unix_ns, not both",
269            ));
270        }
271        (Some(s), None) => Some(When::In(secs(s)?)),
272        (None, Some(ns)) => Some(When::At(unix("start_unix_ns", ns)?)),
273        (None, None) => None,
274    };
275    let on_miss = miss_policy(spec.on_miss)?;
276    if let Some(w) = start {
277        t = match w {
278            When::In(d) => t.start_in(d),
279            When::At(at) => t.start_at(at),
280        };
281        if let Some(tol) = spec.tolerance_s {
282            t = t.start_tolerance(secs(tol)?);
283        }
284        t = t.on_miss(on_miss);
285    } else if spec.tolerance_s.is_some() {
286        return Err(Status::invalid_argument(
287            "tolerance_s requires a start time",
288        ));
289    }
290    if let Some(d) = spec.duration_hint_s {
291        t = t.duration_hint(secs(d)?);
292    }
293    if let Some(d) = spec.timeout_s {
294        t = t.timeout(secs(d)?);
295    }
296    if let Some(r) = &spec.retry {
297        t = t.retry(retry(r)?);
298    }
299    for d in spec.deps {
300        t = t.dep(Dep {
301            target: task_ref(d.target, limits)?,
302            cond: edge_cond(d.cond)?,
303        });
304    }
305    let mut resources: Vec<_> = spec.resources.into_iter().collect();
306    resources.sort();
307    for (name, amount) in resources {
308        check_str("resource name", &name, 64)?;
309        t = t.resource(name, amount);
310    }
311    let mut tags: Vec<_> = spec.tags.into_iter().collect();
312    tags.sort();
313    for (k, v) in tags {
314        check_str("tag key", &k, limits.max_tag_len)?;
315        check_str("tag value", &v, limits.max_tag_len)?;
316        if k.starts_with("cronoq.") {
317            return Err(Status::invalid_argument(
318                "tag keys starting with 'cronoq.' are reserved",
319            ));
320        }
321        t = t.tag(k, v);
322    }
323    check_str("tenant", &spec.tenant, limits.max_key_len)?;
324    if !spec.tenant.is_empty() {
325        t = t.tenant(spec.tenant);
326    }
327    let mut labels: Vec<_> = spec.labels.into_iter().collect();
328    labels.sort();
329    for (k, v) in labels {
330        check_label(&k, &v, limits)?;
331        t = t.label(k, v);
332    }
333    if !spec.content_type.is_empty() {
334        t = t.tag(CONTENT_TYPE_TAG, spec.content_type);
335    }
336    Ok(t)
337}
338
339/// Check one worker label: a non-empty name and a value within the tag length limits.
340pub fn check_label(name: &str, value: &str, limits: &LimitsConfig) -> Result<(), Status> {
341    check_str("label name", name, limits.max_tag_len)?;
342    check_str("label value", value, limits.max_tag_len)?;
343    if name.is_empty() {
344        return Err(Status::invalid_argument("label names must not be empty"));
345    }
346    Ok(())
347}
348
349/// Worker labels from the wire, checked and sorted.
350pub fn worker_labels(
351    labels: std::collections::HashMap<String, String>,
352    limits: &LimitsConfig,
353) -> Result<cronoq::Labels, Status> {
354    if labels.len() > limits.max_tags {
355        return Err(Status::invalid_argument(format!(
356            "at most {} labels are allowed",
357            limits.max_tags
358        )));
359    }
360    let mut v = Vec::with_capacity(labels.len());
361    for (k, val) in labels {
362        check_label(&k, &val, limits)?;
363        v.push((
364            std::sync::Arc::<str>::from(k),
365            std::sync::Arc::<str>::from(val),
366        ));
367    }
368    v.sort();
369    Ok(v.into())
370}
371
372/// Map an engine status to the wire.
373pub fn status(s: TaskStatus) -> pb::TaskStatus {
374    match s {
375        TaskStatus::Blocked => pb::TaskStatus::Blocked,
376        TaskStatus::Scheduled => pb::TaskStatus::Scheduled,
377        TaskStatus::Ready => pb::TaskStatus::Ready,
378        TaskStatus::Running => pb::TaskStatus::Running,
379        TaskStatus::Retrying => pb::TaskStatus::Retrying,
380        TaskStatus::Succeeded => pb::TaskStatus::Succeeded,
381        TaskStatus::Failed => pb::TaskStatus::Failed,
382        TaskStatus::DeadLettered => pb::TaskStatus::DeadLettered,
383        TaskStatus::Cancelled => pb::TaskStatus::Cancelled,
384        TaskStatus::Skipped => pb::TaskStatus::Skipped,
385        TaskStatus::Missed => pb::TaskStatus::Missed,
386        TaskStatus::Shed => pb::TaskStatus::Shed,
387    }
388}
389
390/// Map a wire status to the engine's; `UNSPECIFIED` and unknown values are refused.
391pub fn status_from_pb(s: i32) -> Result<TaskStatus, Status> {
392    Ok(match pb::TaskStatus::try_from(s) {
393        Ok(pb::TaskStatus::Blocked) => TaskStatus::Blocked,
394        Ok(pb::TaskStatus::Scheduled) => TaskStatus::Scheduled,
395        Ok(pb::TaskStatus::Ready) => TaskStatus::Ready,
396        Ok(pb::TaskStatus::Running) => TaskStatus::Running,
397        Ok(pb::TaskStatus::Retrying) => TaskStatus::Retrying,
398        Ok(pb::TaskStatus::Succeeded) => TaskStatus::Succeeded,
399        Ok(pb::TaskStatus::Failed) => TaskStatus::Failed,
400        Ok(pb::TaskStatus::DeadLettered) => TaskStatus::DeadLettered,
401        Ok(pb::TaskStatus::Cancelled) => TaskStatus::Cancelled,
402        Ok(pb::TaskStatus::Skipped) => TaskStatus::Skipped,
403        Ok(pb::TaskStatus::Missed) => TaskStatus::Missed,
404        Ok(pb::TaskStatus::Shed) => TaskStatus::Shed,
405        Ok(pb::TaskStatus::Unspecified) | Err(_) => {
406            return Err(Status::invalid_argument(format!("unknown task status {s}")));
407        }
408    })
409}
410
411/// Map an overlap policy to the wire.
412pub fn overlap_to_pb(o: OverlapPolicy) -> pb::Overlap {
413    match o {
414        OverlapPolicy::Skip => pb::Overlap::Skip,
415        OverlapPolicy::Allow => pb::Overlap::Allow,
416        OverlapPolicy::Queue => pb::Overlap::Queue,
417        OverlapPolicy::QueueOnSuccess => pb::Overlap::QueueOnSuccess,
418    }
419}
420
421/// Map an engine status event to the wire.
422pub fn event(e: &StatusEvent, pools: &[Arc<str>], clock: &Clock) -> pb::StatusEvent {
423    pb::StatusEvent {
424        id: e.id.0,
425        pool: pools
426            .get(e.pool.0 as usize)
427            .map(|p| p.to_string())
428            .unwrap_or_default(),
429        key: e.key.as_deref().unwrap_or_default().to_string(),
430        status: status(e.status) as i32,
431        attempt: e.attempt,
432        at_unix_ns: unix_ns(clock.to_system(e.at)),
433        message: e.message.as_deref().unwrap_or_default().to_string(),
434    }
435}
436
437/// Map a scheduler snapshot to the wire.
438pub fn snapshot(s: &Snapshot, workers: &dyn Fn(&str) -> usize) -> pb::SnapshotResponse {
439    let t = &s.totals;
440    let pools: Vec<pb::PoolSnapshot> = s
441        .pools
442        .iter()
443        .map(|p| pb::PoolSnapshot {
444            name: p.name.to_string(),
445            capacity: p.capacity,
446            busy: p.busy,
447            held: p.held,
448            desired: p.desired,
449            pending_by_class: p.pending_by_class.iter().map(|v| *v as u64).collect(),
450            reservations: p.reservations as u64,
451            dispatched: p.dispatched,
452            succeeded: p.succeeded,
453            failed: p.failed,
454            priority_adherence: p.priority_adherence,
455            fixed_start_hit_rate: p.fixed_start_hit_rate,
456            utilization: p.utilization,
457            schedule_efficiency: p.schedule_efficiency,
458            wait_p50_s: p.wait_p50.as_secs_f64(),
459            wait_p90_s: p.wait_p90.as_secs_f64(),
460            wait_p99_s: p.wait_p99.as_secs_f64(),
461            cost: p.cost,
462            workers_connected: workers(&p.name) as u64,
463            groups: p
464                .groups
465                .iter()
466                .map(|g| pb::WorkerGroup {
467                    labels: g
468                        .labels
469                        .iter()
470                        .map(|(k, v)| (k.to_string(), v.to_string()))
471                        .collect(),
472                    capacity: g.capacity,
473                    busy: g.busy,
474                })
475                .collect(),
476        })
477        .collect();
478    pb::SnapshotResponse {
479        workers_connected: pools.iter().map(|p| p.workers_connected).sum(),
480        pools,
481        live_tasks: s.live_tasks as u64,
482        running: s.running as u64,
483        pending: s.pending as u64,
484        totals: Some(pb::Totals {
485            submitted: t.submitted,
486            dispatched: t.dispatched,
487            succeeded: t.succeeded,
488            failed: t.failed,
489            dead_lettered: t.dead_lettered,
490            cancelled: t.cancelled,
491            skipped: t.skipped,
492            missed: t.missed,
493            shed: t.shed,
494            retries: t.retries,
495            timeouts: t.timeouts,
496            lease_expiries: t.lease_expiries,
497            stale_completions: t.stale_completions,
498        }),
499        unavailable_shards: Vec::new(),
500    }
501}
502
503/// Map a scheduler error to a gRPC status. Store details are logged, not returned.
504pub fn error(e: Error) -> Status {
505    match e {
506        Error::Submit(e) => submit_error(e),
507        Error::Shutdown => Status::unavailable("scheduler is shutting down"),
508        Error::Config(m) => Status::invalid_argument(m),
509        Error::Store(e) => {
510            tracing::warn!(error = %e, "store error");
511            Status::unavailable("store unavailable")
512        }
513        e => Status::internal(e.to_string()),
514    }
515}
516
517/// Map a submission error to a gRPC status.
518pub fn submit_error(e: SubmitError) -> Status {
519    let msg = e.to_string();
520    match e {
521        SubmitError::Full | SubmitError::Overloaded => Status::resource_exhausted(msg),
522        SubmitError::ReservationConflict => Status::failed_precondition(msg),
523        SubmitError::UnknownPool(_)
524        | SubmitError::UnknownResource(_)
525        | SubmitError::UnknownDependency(_) => Status::not_found(msg),
526        _ => Status::invalid_argument(msg),
527    }
528}
529
530#[cfg(test)]
531mod tests {
532    use super::*;
533
534    fn lim() -> LimitsConfig {
535        LimitsConfig::default()
536    }
537
538    #[test]
539    fn durations_are_validated() {
540        for bad in [-1.0, f64::NAN, f64::INFINITY, 1e300] {
541            assert!(secs(bad).is_err(), "{bad}");
542        }
543        assert_eq!(secs(1.5).unwrap(), Duration::from_millis(1500));
544    }
545
546    #[test]
547    fn task_from_proto() {
548        let spec = pb::TaskSpec {
549            payload: b"hi".to_vec(),
550            content_type: "text/plain".into(),
551            kind: "k".into(),
552            pool: "p".into(),
553            priority: pb::Priority::High as i32,
554            key: "key".into(),
555            deadline_in_s: Some(5.0),
556            start_in_s: Some(10.0),
557            tolerance_s: Some(2.0),
558            timeout_s: Some(1.5),
559            deps: vec![pb::Dep {
560                target: Some(pb::TaskRef {
561                    target: Some(pb::task_ref::Target::Key("a".into())),
562                }),
563                cond: pb::EdgeCond::OnFailure as i32,
564            }],
565            resources: [("gpu".to_string(), 1)].into_iter().collect(),
566            ..Default::default()
567        };
568        let t = task(spec, &lim()).unwrap();
569        let s = t.spec();
570        assert_eq!(&*s.kind, "k");
571        assert_eq!(&*s.pool, "p");
572        assert_eq!(s.priority, Priority::High);
573        assert_eq!(s.key.as_deref(), Some("key"));
574        assert_eq!(s.timeout, Some(Duration::from_millis(1500)));
575        assert_eq!(s.deps[0].cond, EdgeCond::OnFailure);
576        assert_eq!(s.resources[0].1, 1);
577        assert!(
578            s.tags
579                .iter()
580                .any(|(k, v)| &**k == CONTENT_TYPE_TAG && &**v == "text/plain")
581        );
582    }
583
584    #[test]
585    fn rejects_hostile_specs() {
586        let bad = [
587            pb::TaskSpec {
588                timeout_s: Some(-1.0),
589                ..Default::default()
590            },
591            pb::TaskSpec {
592                key: "evil\0".into(),
593                ..Default::default()
594            },
595            pb::TaskSpec {
596                kind: "k\0".into(),
597                ..Default::default()
598            },
599            pb::TaskSpec {
600                pool: "p\0".into(),
601                ..Default::default()
602            },
603            pb::TaskSpec {
604                tags: [("a".to_string(), "\0".to_string())].into_iter().collect(),
605                ..Default::default()
606            },
607            pb::TaskSpec {
608                resources: [("r\0".to_string(), 1)].into_iter().collect(),
609                ..Default::default()
610            },
611            pb::TaskSpec {
612                key: "k".repeat(1000),
613                ..Default::default()
614            },
615            pb::TaskSpec {
616                deadline_unix_ns: Some(-5),
617                ..Default::default()
618            },
619            pb::TaskSpec {
620                on_miss: 99,
621                start_in_s: Some(1.0),
622                ..Default::default()
623            },
624            pb::TaskSpec {
625                deps: vec![pb::Dep {
626                    target: Some(pb::TaskRef {
627                        target: Some(pb::task_ref::Target::Id(1)),
628                    }),
629                    cond: 42,
630                }],
631                ..Default::default()
632            },
633            pb::TaskSpec {
634                tags: (0..100).map(|i| (format!("t{i}"), String::new())).collect(),
635                ..Default::default()
636            },
637            pb::TaskSpec {
638                tags: [("cronoq.content_type".to_string(), "x".to_string())]
639                    .into_iter()
640                    .collect(),
641                ..Default::default()
642            },
643            pb::TaskSpec {
644                retry: Some(pb::RetryPolicy {
645                    max_attempts: 1,
646                    multiplier: f64::NAN,
647                    ..Default::default()
648                }),
649                ..Default::default()
650            },
651        ];
652        for (i, s) in bad.into_iter().enumerate() {
653            assert!(task(s, &lim()).is_err(), "case {i} should be rejected");
654        }
655    }
656
657    #[test]
658    fn sanitizes_worker_text() {
659        assert_eq!(sanitize_message("a\0b"), "a\u{FFFD}b");
660        assert!(sanitize_message(&"é".repeat(10_000)).len() <= MAX_MESSAGE_LEN);
661        assert_eq!(
662            sanitize_name("w\n\x1b[31mERROR admin token leaked"),
663            "w[31mERROR admin token leaked"
664        );
665    }
666}
cronoq_server/convert.rs

cronoq_server/
convert.rs
