Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Change-event bus.
3 : //!
4 : //! The event and its transport. `bus = local`, the default and what the
5 : //! ETSI pipeline runs, carries changes in-process through the store's change
6 : //! hook and needs nothing from here but `ChangeEvent`. `bus = nats` adds the
7 : //! JetStream spine (`ANTARES_CHANGES`, durable pull consumers, KV
8 : //! subscription mirror) that makes multi-instance roles possible, and
9 : //! becomes mandatory only on scale-out. The composition root
10 : //! (`antares-broker/src/wiring.rs`) is the only place that names either.
11 : #![cfg_attr(not(test), warn(clippy::expect_used))]
12 :
13 : pub mod nats;
14 : pub mod subjects;
15 :
16 : use antares_model::{EntityId, TenantId};
17 : use serde::{Deserialize, Serialize};
18 :
19 : /// Operation kind — mirrors Scorpio's requestType int registry as an enum.
20 : #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
21 : #[serde(rename_all = "camelCase")]
22 : pub enum ChangeOp {
23 : Create,
24 : Update,
25 : Append,
26 : Merge,
27 : Replace,
28 : Delete,
29 : BatchCreate,
30 : BatchUpsert,
31 : BatchUpdate,
32 : BatchDelete,
33 : BatchMerge,
34 : }
35 :
36 : /// Claim-check reference: events whose payload exceeds
37 : /// [`CLAIM_CHECK_BYTES`] carry this instead of the inline body. NATS caps
38 : /// messages at ~1 MB and Antares never chunks, so the body travels out of
39 : /// band: the publisher keeps the outbox row that holds the whole event, and
40 : /// the consumer reads it back by the event's `seq`. The store's current row
41 : /// is the after-image and can stand in for `payload` alone — a before-image
42 : /// is not derivable from it, which is why the reference is not a document
43 : /// lookup.
44 : #[derive(Debug, Clone, Serialize, Deserialize)]
45 : pub struct PayloadRef {
46 : pub entity_id: EntityId,
47 : pub version: i64,
48 : }
49 :
50 : /// Inline-payload ceiling before the claim check kicks in (256 KB).
51 : pub const CLAIM_CHECK_BYTES: usize = 256 * 1024;
52 :
53 : /// One entity change. Self-contained: carries payload AND prev_payload so
54 : /// consumers (matcher, temporal recorder) never re-read the DB per event.
55 : /// `version` is the entity row version bumped under the write lock —
56 : /// state-projecting consumers apply last-writer-wins on
57 : /// `(incarnation, version)`; `incarnation` is the row's created_at, which
58 : /// disambiguates delete/recreate (the version restarts at 1).
59 : #[derive(Debug, Clone, Serialize, Deserialize)]
60 : pub struct ChangeEvent {
61 : pub tenant: TenantId,
62 : pub entity_id: EntityId,
63 : pub types: Vec<String>,
64 : pub op: ChangeOp,
65 : pub changed_attrs: Vec<String>,
66 : pub payload: Option<serde_json::Value>,
67 : pub prev_payload: Option<serde_json::Value>,
68 : pub version: i64,
69 : /// The row's created_at — the incarnation half of the ordering key.
70 : #[serde(default)]
71 : pub incarnation: String,
72 : /// Outbox row id — the `Nats-Msg-Id` dedup key. 0 = local bus.
73 : #[serde(default)]
74 : pub seq: i64,
75 : /// Claim-check: set when `payload` was stripped for size.
76 : #[serde(default, skip_serializing_if = "Option::is_none")]
77 : pub payload_ref: Option<PayloadRef>,
78 : #[serde(default, skip_serializing_if = "Option::is_none")]
79 : pub prev_payload_ref: Option<PayloadRef>,
80 : }
81 :
82 : /// A body's serialized size against the claim-check ceiling. An
83 : /// unserializable body counts as zero: it cannot be published either way, and
84 : /// treating it as oversized would retain a row nothing can resolve.
85 34 : fn over(v: &Option<serde_json::Value>, limit: usize) -> bool {
86 34 : v.as_ref()
87 34 : .is_some_and(|p| serde_json::to_vec(p).map(|b| b.len()).unwrap_or(0) > limit)
88 34 : }
89 :
90 : impl ChangeEvent {
91 : /// True when [`ChangeEvent::claim_check`] would strip a body at `limit`.
92 : /// The publisher asks before it publishes: a stripped body has to stay
93 : /// readable somewhere the consumer can reach, and the drain is the last
94 : /// holder of the whole event.
95 6 : pub fn claim_checked_at(&self, limit: usize) -> bool {
96 6 : over(&self.payload, limit) || over(&self.prev_payload, limit)
97 6 : }
98 :
99 : /// Claim-check: replace any inline body over `limit` bytes with a
100 : /// reference. Oversized entities are rare; the common path is untouched.
101 12 : pub fn claim_check(mut self, limit: usize) -> Self {
102 12 : if over(&self.payload, limit) {
103 4 : self.payload = None;
104 4 : self.payload_ref = Some(PayloadRef {
105 4 : entity_id: self.entity_id.clone(),
106 4 : version: self.version,
107 4 : });
108 8 : }
109 12 : if over(&self.prev_payload, limit) {
110 6 : self.prev_payload = None;
111 6 : self.prev_payload_ref = Some(PayloadRef {
112 6 : entity_id: self.entity_id.clone(),
113 6 : // saturating: a decoded event's version is whatever the wire
114 6 : // said, and wrapping would reference the wrong document
115 6 : version: self.version.saturating_sub(1),
116 6 : });
117 6 : }
118 : // The ENVELOPE must fit too, or the publish is refused by the bus and
119 : // the outbox drain retries the same row forever. changed_attrs is the
120 : // one member that scales with the entity's width, and no consumer
121 : // reads it off the wire (process_change re-derives the diff from the
122 : // payloads); types stay — the publish subject is built from them.
123 12 : if serde_json::to_vec(&self)
124 12 : .map(|b| b.len())
125 12 : .unwrap_or(usize::MAX)
126 12 : > limit
127 2 : {
128 2 : self.changed_attrs = Vec::new();
129 10 : }
130 12 : self
131 12 : }
132 : }
133 :
134 : #[cfg(test)]
135 : mod tests {
136 : use super::*;
137 :
138 10 : fn event(version: i64) -> ChangeEvent {
139 10 : ChangeEvent {
140 10 : tenant: TenantId::default(),
141 10 : entity_id: EntityId::new("urn:ngsi-ld:Vehicle:A1").expect("valid urn"),
142 10 : types: vec!["https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()],
143 10 : op: ChangeOp::Create,
144 10 : changed_attrs: vec![],
145 10 : payload: Some(serde_json::json!({"speed": 80})),
146 10 : prev_payload: None,
147 10 : version,
148 10 : incarnation: "2026-08-05T00:00:00Z".into(),
149 10 : seq: 0,
150 10 : payload_ref: None,
151 10 : prev_payload_ref: None,
152 10 : }
153 10 : }
154 :
155 : #[test]
156 2 : fn event_round_trips_through_serde() {
157 2 : let e = event(7);
158 2 : let bytes = serde_json::to_vec(&e).expect("serialize");
159 2 : let back: ChangeEvent = serde_json::from_slice(&bytes).expect("deserialize");
160 2 : assert_eq!(back.version, 7);
161 2 : assert_eq!(back.op, ChangeOp::Create);
162 2 : assert_eq!(back.incarnation, "2026-08-05T00:00:00Z");
163 2 : }
164 :
165 : #[test]
166 2 : fn claim_check_bounds_the_envelope_not_only_the_bodies() {
167 2 : let mut e = event(9);
168 2 : e.payload = Some(serde_json::json!({"small": true}));
169 : // the envelope itself outgrows the limit: thousands of attribute
170 : // IRIs from one wide entity, with both bodies tiny
171 2 : e.changed_attrs = (0..20_000)
172 40000 : .map(|i| format!("https://example.org/ngsi-ld/attributes/generated/a{i:05}"))
173 2 : .collect();
174 2 : let limit = 256 * 1024;
175 2 : let checked = e.claim_check(limit);
176 2 : let wire = serde_json::to_vec(&checked).expect("serialize");
177 2 : assert!(
178 2 : wire.len() <= limit,
179 : "the published message must fit the bus limit, got {} bytes",
180 0 : wire.len()
181 : );
182 2 : assert!(
183 2 : checked.payload.is_some(),
184 : "a small body is not the thing to strip for an oversized envelope"
185 : );
186 2 : assert!(
187 2 : checked.changed_attrs.is_empty(),
188 : "changed_attrs is re-derived by the consumer from the payloads, so it goes first"
189 : );
190 2 : assert!(
191 2 : !checked.types.is_empty(),
192 : "types must survive — the publish subject is built from them"
193 : );
194 2 : }
195 :
196 : /// The publisher asks `claim_checked_at` BEFORE it publishes and keeps
197 : /// the outbox row when the answer is yes. An answer that disagrees with
198 : /// the strip either keeps every row (the outbox never drains) or keeps
199 : /// none (the consumer resolves a reference to a row that is gone).
200 : #[test]
201 2 : fn claim_checked_at_answers_for_the_bodies_the_strip_takes() {
202 2 : let mut fits = event(3);
203 2 : fits.payload = Some(serde_json::json!({"small": true}));
204 2 : fits.prev_payload = Some(serde_json::json!({"small": false}));
205 2 : assert!(!fits.claim_checked_at(512));
206 2 : assert!(fits.clone().claim_check(512).payload_ref.is_none());
207 :
208 2 : let mut prev_only = fits.clone();
209 2 : prev_only.prev_payload = Some(serde_json::Value::String("x".repeat(1024)));
210 2 : assert!(
211 2 : prev_only.claim_checked_at(512),
212 : "a stripped before-image is the one the store cannot answer for"
213 : );
214 2 : assert!(prev_only.claim_check(512).prev_payload_ref.is_some());
215 :
216 2 : let mut both = fits;
217 2 : both.payload = Some(serde_json::Value::String("x".repeat(1024)));
218 2 : both.prev_payload = Some(serde_json::Value::String("y".repeat(1024)));
219 2 : assert!(both.claim_checked_at(512));
220 2 : }
221 :
222 : #[test]
223 2 : fn claim_check_strips_only_oversized_bodies() {
224 2 : let mut e = event(3);
225 2 : e.prev_payload = Some(serde_json::json!({"small": true}));
226 2 : e.payload = Some(serde_json::Value::String("x".repeat(1024)));
227 2 : let checked = e.claim_check(512);
228 2 : assert!(checked.payload.is_none(), "over-limit body stripped");
229 2 : let r = checked.payload_ref.as_ref().expect("ref set");
230 2 : assert_eq!(r.version, 3);
231 2 : assert!(
232 2 : checked.prev_payload.is_some() && checked.prev_payload_ref.is_none(),
233 : "under-limit body inline"
234 : );
235 2 : }
236 :
237 : /// A decoded event carries whatever `version` the wire said. The claim
238 : /// check derives the previous version from it, and that arithmetic must
239 : /// not overflow — an overflow is a panic in debug builds and a wrap in
240 : /// release, i.e. a reference to the wrong document.
241 : #[test]
242 2 : fn claim_check_does_not_underflow_on_an_extreme_version() {
243 2 : let mut e = event(i64::MIN);
244 2 : e.prev_payload = Some(serde_json::Value::String("x".repeat(1024)));
245 2 : let checked = e.claim_check(512);
246 2 : assert_eq!(
247 2 : checked.prev_payload_ref.expect("ref set").version,
248 : i64::MIN,
249 : "must saturate, never wrap to i64::MAX"
250 : );
251 2 : }
252 : }
|
