1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
//! Channel schema resolution: turns a configured channel's loaded schema (or the
//! permissive empty default) into the JSON Schema document the channel is built
//! with and the protocol schema id advertised at subscribe time.
use liminal::protocol::SchemaId as ProtocolSchemaId;
use serde_json::Value;
use crate::config::types::ChannelDef;
/// A resolved channel schema: the JSON Schema document plus the protocol schema id
/// advertised to subscribers.
pub(super) struct ChannelSchema {
/// JSON Schema document fed to the channel's validation engine.
pub document: Value,
/// Protocol schema id derived from the schema bytes, advertised on subscribe.
pub protocol_id: ProtocolSchemaId,
}
/// Canonical bytes of the permissive empty schema. A schema-less channel derives
/// its protocol id from these bytes so the id is stable across restarts.
const EMPTY_SCHEMA_BYTES: &[u8] = b"{}";
/// Resolves the JSON Schema document and protocol schema id for a channel.
///
/// A channel whose `schema_ref` was loaded during config validation uses that
/// parsed document and derives its protocol id from the RAW loaded schema bytes. A
/// channel with no loaded schema (either `schema_ref: None`, or a config that was
/// never validated) keeps the permissive empty schema `{}` that accepts any JSON
/// payload, deriving its protocol id from the `{}` bytes.
pub(super) fn resolve_channel_schema(channel: &ChannelDef) -> ChannelSchema {
// A DECLARED schema_ref with no loaded schema means this config skipped
// validation (which loads the ref or fails): the channel silently runs
// permissive against the operator's declared intent, so say so. The
// legitimate `schema_ref: None` case stays silent, and resolution itself is
// unchanged either way.
if channel.schema_ref.is_some() && channel.loaded_schema.is_none() {
tracing::warn!(
channel = %channel.name,
"channel declares a schema_ref but no schema was loaded; running permissive (config skipped validation)"
);
}
channel.loaded_schema.as_ref().map_or_else(
|| ChannelSchema {
document: Value::Object(serde_json::Map::new()),
protocol_id: schema_id_from_bytes(EMPTY_SCHEMA_BYTES),
},
|loaded| ChannelSchema {
document: loaded.document.clone(),
protocol_id: schema_id_from_bytes(&loaded.bytes),
},
)
}
/// Resolves the JSON Schema document and protocol schema id from RAW schema
/// bytes.
///
/// This is [`resolve_channel_schema`]'s two branches lifted off [`ChannelDef`]:
/// `None` is the permissive empty schema `{}` with the id derived from the `{}`
/// bytes, and `Some(bytes)` is the parsed document with the id derived from
/// those same bytes. The only difference is WHO parses — config validation
/// already parsed a boot channel's file into `loaded_schema.document`, while a
/// runtime registration hands over bytes and this function parses them — so a
/// parse failure is surfaced here instead of at config-load time.
///
/// The boot path deliberately keeps calling [`resolve_channel_schema`]: routing
/// it through here would re-parse a document config validation already parsed,
/// which is a behaviour change (a different error site) in a step that claims
/// the boot path is untouched.
///
/// # Errors
/// Returns the `serde_json` parse error when `schema_bytes` is not valid JSON.
pub(super) fn resolve_schema_bytes(
schema_bytes: Option<&[u8]>,
) -> Result<ChannelSchema, serde_json::Error> {
let Some(bytes) = schema_bytes else {
return Ok(ChannelSchema {
document: Value::Object(serde_json::Map::new()),
protocol_id: schema_id_from_bytes(EMPTY_SCHEMA_BYTES),
});
};
Ok(ChannelSchema {
document: serde_json::from_slice(bytes)?,
protocol_id: schema_id_from_bytes(bytes),
})
}
/// The protocol schema id for raw schema bytes: liminal's one public
/// derivation, [`ProtocolSchemaId::from_schema_bytes`]. This was a private
/// copy of the same FNV-1a spread until the derivation went public; it stays
/// as the single name this module reaches it by.
fn schema_id_from_bytes(schema_bytes: &[u8]) -> ProtocolSchemaId {
ProtocolSchemaId::from_schema_bytes(schema_bytes)
}
#[cfg(test)]
mod tests {
use super::{resolve_channel_schema, resolve_schema_bytes, schema_id_from_bytes};
use crate::config::types::{ChannelDef, LoadedSchema};
fn channel(schema_ref: Option<&str>, loaded: Option<LoadedSchema>) -> ChannelDef {
ChannelDef {
name: "orders".to_owned(),
schema_ref: schema_ref.map(Into::into),
durable: false,
loaded_schema: loaded,
}
}
#[test]
fn schema_ids_are_deterministic_and_content_addressed() {
assert_eq!(schema_id_from_bytes(b"a"), schema_id_from_bytes(b"a"));
assert_ne!(schema_id_from_bytes(b"a"), schema_id_from_bytes(b"b"));
}
#[test]
fn loaded_schema_id_derives_from_loaded_bytes() {
let bytes = br#"{"type":"object"}"#.to_vec();
let document = serde_json::json!({"type": "object"});
let resolved = resolve_channel_schema(&channel(
Some("orders.json"),
Some(LoadedSchema {
bytes: bytes.clone(),
document: document.clone(),
}),
));
assert_eq!(resolved.document, document);
assert_eq!(resolved.protocol_id, schema_id_from_bytes(&bytes));
}
#[test]
fn schema_less_channel_is_permissive_and_uses_empty_schema_id() {
let resolved = resolve_channel_schema(&channel(None, None));
assert_eq!(resolved.document, serde_json::json!({}));
assert_eq!(resolved.protocol_id, schema_id_from_bytes(b"{}"));
}
/// The raw-bytes lift resolves the SAME two branches as the `ChannelDef`
/// form: absent bytes are the permissive `{}` with the `{}` id, and present
/// bytes give the parsed document with the id derived from those bytes. A
/// runtime registration and a boot channel over identical schema bytes must
/// therefore land on identical entries.
#[test]
fn raw_bytes_resolution_matches_the_channel_def_form() -> Result<(), serde_json::Error> {
let empty = resolve_schema_bytes(None)?;
let from_def = resolve_channel_schema(&channel(None, None));
assert_eq!(empty.document, from_def.document);
assert_eq!(empty.protocol_id, from_def.protocol_id);
let bytes = br#"{"type":"object"}"#.to_vec();
let loaded = resolve_schema_bytes(Some(&bytes))?;
let loaded_from_def = resolve_channel_schema(&channel(
Some("orders.json"),
Some(LoadedSchema {
bytes,
document: serde_json::json!({"type": "object"}),
}),
));
assert_eq!(loaded.document, loaded_from_def.document);
assert_eq!(loaded.protocol_id, loaded_from_def.protocol_id);
Ok(())
}
/// Bytes that are not JSON at all are refused HERE, by the parser, rather
/// than reaching schema compilation with a half-built document.
#[test]
fn raw_bytes_resolution_refuses_bytes_that_are_not_json() {
assert!(resolve_schema_bytes(Some(b"not json")).is_err());
}
/// A channel that DECLARED a `schema_ref` but has no loaded schema is a config
/// that skipped validation, silently running permissive — that must be
/// warned about, naming the channel. Resolution itself stays permissive.
#[test]
fn declared_but_unloaded_schema_warns_naming_the_channel() {
let log = crate::test_log::CapturedLog::default();
let resolved = log.capture(|| resolve_channel_schema(&channel(Some("orders.json"), None)));
assert_eq!(resolved.document, serde_json::json!({}));
let output = log.contents();
assert!(
output.contains("WARN") && output.contains("orders"),
"expected a warn naming the channel, got: {output}"
);
}
/// Control: the legitimate `schema_ref: None` case stays silent.
#[test]
fn schema_less_channel_resolves_without_warning() {
let log = crate::test_log::CapturedLog::default();
let _resolved = log.capture(|| resolve_channel_schema(&channel(None, None)));
let output = log.contents();
assert!(
!output.contains("WARN"),
"a schema-less channel must resolve silently, got: {output}"
);
}
}