aion/loader/
declared_queues.rs

1//! What one read of the deployed contracts could — and could not — say about
2//! declared task queues.
3//!
4//! A catalog can retain entries whose stored identity carries no decodable
5//! contract at all: every package deployed before the `.v4` contract identity
6//! reads that way. The handshake migration creates exactly that mixture —
7//! old packages retained beside newly deployed `.v4` ones — so a bare set of
8//! found queues is not an honest answer. Such a set reads as authoritative
9//! while silently omitting whatever the undecodable entries declare, and a
10//! caller that treats its absences as facts refuses dispatches nobody
11//! contradicted.
12//!
13//! This type keeps both halves of the answer together, so a reader can tell
14//! "no deployed contract declares this queue" apart from "this read could not
15//! have seen it".
16
17use std::collections::BTreeSet;
18
19use aion_package::ContractIdentityError;
20
21use super::load::LoadedWorkflow;
22
23/// The queues one catalog read positively found, with the stored identities
24/// that same read could not decode.
25///
26/// A positive find is a fact regardless of what the read missed. An ABSENCE is
27/// only a fact when [`Self::covers_every_entry`] holds.
28#[derive(Clone, Debug, Default, Eq, PartialEq)]
29pub struct DeclaredQueues {
30    declared: BTreeSet<String>,
31    undecodable_identities: Vec<String>,
32}
33
34impl DeclaredQueues {
35    /// Records one read: the queues found, and the stored identities that could
36    /// not be decoded — empty exactly when the read covered every entry.
37    #[must_use]
38    pub fn new(declared: BTreeSet<String>, undecodable_identities: Vec<String>) -> Self {
39        Self {
40            declared,
41            undecodable_identities,
42        }
43    }
44
45    /// Whether a decoded contract declares `task_queue`.
46    #[must_use]
47    pub fn declares(&self, task_queue: &str) -> bool {
48        self.declared.contains(task_queue)
49    }
50
51    /// Whether every retained entry decoded — the only condition under which
52    /// this read's absences carry information.
53    #[must_use]
54    pub fn covers_every_entry(&self) -> bool {
55        self.undecodable_identities.is_empty()
56    }
57
58    /// The stored identities this read could not decode, in catalog order.
59    #[must_use]
60    pub fn undecodable_identities(&self) -> &[String] {
61        &self.undecodable_identities
62    }
63
64    /// Whether the read found no queue declaration at all. Such a catalog
65    /// contradicts nothing, whether or not the read was complete.
66    #[must_use]
67    pub fn found_no_declaration(&self) -> bool {
68        self.declared.is_empty()
69    }
70
71    /// Every queue this read found.
72    #[must_use]
73    pub fn declared(&self) -> &BTreeSet<String> {
74        &self.declared
75    }
76
77    /// Reads the declarations out of retained catalog entries.
78    ///
79    /// This is the site that OBSERVES an entry it cannot decode, so this is the
80    /// site that says so: an undecodable entry is recorded by its stored
81    /// identity and reported at WARN, naming what must be re-deployed. It is
82    /// never skipped in silence — a skipped entry turns into a confident,
83    /// terminal refusal three seams downstream.
84    pub(crate) fn read<'entries>(
85        entries: impl IntoIterator<Item = &'entries LoadedWorkflow>,
86    ) -> Self {
87        let mut declared = BTreeSet::new();
88        let mut undecodable_identities = Vec::new();
89        for workflow in entries {
90            match workflow.contract() {
91                Ok(contract) => {
92                    for worker in &contract.workers {
93                        declared.insert(worker.task_queue.clone());
94                    }
95                }
96                Err(ContractIdentityError::RedeployRequired { stored_version }) => {
97                    undecodable_identities.push(stored_version);
98                }
99            }
100        }
101        if !undecodable_identities.is_empty() {
102            tracing::warn!(
103                undecodable_identities = %undecodable_identities.join(", "),
104                declared_queues = declared.len(),
105                "catalog entries carry no decodable contract; their task queues are unknowable \
106                 and no queue can be reported undeclared until they are re-deployed under `.v4`"
107            );
108        }
109        Self {
110            declared,
111            undecodable_identities,
112        }
113    }
114}
115
116#[cfg(test)]
117#[path = "declared_queues_tests.rs"]
118mod declared_queues_tests;
aion/loader/declared_queues.rs

aion/loader/
declared_queues.rs
