reifydb_core/key/operator/
state.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	borrow::Cow,
6	fmt::{Display, Formatter, Result as FmtResult},
7	ops::Bound,
8};
9
10use reifydb_codec::key::{
11	deserializer::KeyDeserializer,
12	encode_u8,
13	encoded::{EncodedKey, EncodedKeyRange},
14	serializer::KeySerializer,
15};
16use reifydb_macro::KeyCodec;
17use reifydb_value::util::hash::{Hash128, xxh3_128};
18use serde::{Deserialize, Serialize};
19
20use super::super::KeyTag;
21use crate::{
22	common::OperatorClass,
23	interface::catalog::flow::OperatorId,
24	key::{
25		any::{ByteEncoding, Field, KeyFields, RawEncoding, TaggedKey, Width},
26		bound::{TaggedKeyBound, TaggedKeyBoundRange},
27		operator::{
28			keyspace::{
29				KEYSPACES, KeyspaceVisitor, REGISTERED, dispatch,
30				root::{
31					CustomManagedSuffix, CustomUnmanagedSuffix, NodeCounter, NodeCounterKey,
32					NodeCounterKind,
33				},
34				suffix_width_of,
35			},
36			traits::Keyspace,
37		},
38		typed::{BoundedKey, layout::KeyLayout},
39	},
40	metrics::heap::HeapSize,
41	state::typed::{SuffixBytes, typed_key},
42};
43
44#[repr(transparent)]
45#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
46pub struct GroupId([u8; GroupId::WIDTH]);
47
48impl GroupId {
49	pub const WIDTH: usize = 24;
50
51	const HASH_OFFSET: usize = size_of::<u64>();
52
53	const NOT_A_WINDOW: u64 = u64::MAX;
54
55	pub const ROOT: Self = Self([0; Self::WIDTH]);
56
57	pub const FIRST_NON_ROOT: Self = {
58		let mut bytes = [0u8; Self::WIDTH];
59		bytes[Self::WIDTH - 1] = 1;
60		Self(bytes)
61	};
62
63	pub const UNKEYED: Self = {
64		let mut bytes = [0u8; Self::WIDTH];
65		let mut idx = 0;
66		while idx < Self::HASH_OFFSET {
67			bytes[idx] = u8::MAX;
68			idx += 1;
69		}
70		Self(bytes)
71	};
72
73	pub const MIN: Self = Self([u8::MIN; Self::WIDTH]);
74
75	pub const MAX: Self = Self([u8::MAX; Self::WIDTH]);
76
77	pub const fn from_bytes(bytes: [u8; Self::WIDTH]) -> Self {
78		Self(bytes)
79	}
80
81	pub const fn as_bytes(&self) -> &[u8; Self::WIDTH] {
82		&self.0
83	}
84
85	pub fn of(key: &EncodedKey) -> Self {
86		Self::hashed(xxh3_128(key.as_slice()))
87	}
88
89	pub fn hashed(hash: Hash128) -> Self {
90		match hash.0 {
91			0 => Self::FIRST_NON_ROOT,
92			carried => Self::at(Self::NOT_A_WINDOW, carried),
93		}
94	}
95
96	pub fn window(partition: Hash128, window_id: u64) -> Self {
97		assert!(
98			window_id != Self::NOT_A_WINDOW,
99			"window id {window_id} is the sentinel that marks a group as having no window; a window \
100			 minted at it would collide with the join and distinct groups of the same operator"
101		);
102		Self::at(window_id, partition.0)
103	}
104
105	pub fn window_span(window_id: u64) -> (Self, Self) {
106		(Self::at(window_id, u128::MIN), Self::at(window_id, u128::MAX))
107	}
108
109	pub fn window_id(&self) -> Option<u64> {
110		let mut leading = [0u8; size_of::<u64>()];
111		leading.copy_from_slice(&self.0[..Self::HASH_OFFSET]);
112		match !u64::from_be_bytes(leading) {
113			Self::NOT_A_WINDOW => None,
114			window_id => Some(window_id),
115		}
116	}
117
118	pub fn partition(&self) -> Hash128 {
119		let mut hash = [0u8; size_of::<u128>()];
120		hash.copy_from_slice(&self.0[Self::HASH_OFFSET..]);
121		Hash128(u128::from_be_bytes(hash))
122	}
123
124	fn at(window_id: u64, hash: u128) -> Self {
125		let mut bytes = [0u8; Self::WIDTH];
126		bytes[..Self::HASH_OFFSET].copy_from_slice(&(!window_id).to_be_bytes());
127		bytes[Self::HASH_OFFSET..].copy_from_slice(&hash.to_be_bytes());
128		Self(bytes)
129	}
130
131	pub fn is_root(&self) -> bool {
132		*self == Self::ROOT
133	}
134
135	pub fn successor(&self) -> Option<Self> {
136		let mut bytes = self.0;
137		for byte in bytes.iter_mut().rev() {
138			let (stepped, carried) = byte.overflowing_add(1);
139			*byte = stepped;
140			if !carried {
141				return Some(Self(bytes));
142			}
143		}
144		None
145	}
146
147	pub fn predecessor(&self) -> Option<Self> {
148		let mut bytes = self.0;
149		for byte in bytes.iter_mut().rev() {
150			let (stepped, borrowed) = byte.overflowing_sub(1);
151			*byte = stepped;
152			if !borrowed {
153				return Some(Self(bytes));
154			}
155		}
156		None
157	}
158}
159
160impl Display for GroupId {
161	fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
162		for byte in self.0 {
163			write!(f, "{byte:02x}")?;
164		}
165		Ok(())
166	}
167}
168
169impl HeapSize for GroupId {
170	fn heap_size(&self) -> usize {
171		0
172	}
173}
174
175#[derive(Debug, Clone, Default, PartialEq, Eq)]
176pub struct GroupSet(Vec<GroupId>);
177
178impl GroupSet {
179	pub fn new(groups: impl IntoIterator<Item = GroupId>) -> Self {
180		let mut groups: Vec<GroupId> = groups.into_iter().filter(|g| !g.is_root()).collect();
181		groups.sort_unstable();
182		groups.dedup();
183		Self(groups)
184	}
185
186	pub fn contains(&self, group: GroupId) -> bool {
187		self.0.binary_search(&group).is_ok()
188	}
189
190	pub fn as_slice(&self) -> &[GroupId] {
191		&self.0
192	}
193
194	pub fn len(&self) -> usize {
195		self.0.len()
196	}
197
198	pub fn is_empty(&self) -> bool {
199		self.0.is_empty()
200	}
201}
202
203pub fn group_data_of_inner(inner: &[u8]) -> Option<GroupId> {
204	let mut de = KeyDeserializer::from_bytes(inner);
205	let group = GroupId::from_bytes(de.read_fixed().ok()?);
206	let keyspace = KeyspaceId(de.read_u8().ok()?);
207	if !keyspace.is_data() {
208		return None;
209	}
210	inner.starts_with(&group_inner_prefix(group)).then_some(group)
211}
212
213#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
214pub struct KeyspaceId(pub u8);
215
216impl KeyspaceId {
217	pub const HIGHEST_DATA: u8 = 0x2F;
218
219	pub const NODE_COUNTER: Self = Self(0xFF);
220
221	pub const SOURCE_WATERMARK: Self = Self(0xFE);
222
223	pub const TIMER_WHEEL: Self = Self(0xFD);
224
225	pub const TIMER_INDEX: Self = Self(0xFC);
226
227	pub const JOIN_ROW_MAPPING: Self = Self(0xFB);
228
229	pub const GROUP_ROW_MAPPING: Self = Self(0xFA);
230
231	pub const GUEST_ROW_MAPPING: Self = Self(0xF9);
232
233	pub const ACCUMULATOR: Self = Self(0x00);
234
235	pub const BUFFER: Self = Self(0x01);
236
237	pub const RUNNING: Self = Self(0x02);
238
239	pub const EMIT: Self = Self(0x03);
240
241	pub const ROLLING_EXPIRY: Self = Self(0x04);
242
243	pub const COUNT: Self = Self(0x05);
244
245	pub const ROW_INDEX: Self = Self(0x06);
246
247	pub const SESSION: Self = Self(0x07);
248
249	pub const ROLLING_META: Self = Self(0x08);
250
251	pub const ENGINE_META: Self = Self(0x09);
252
253	pub const DISTINCT_ENTRY: Self = Self(0x0A);
254
255	pub const WINDOW_META: Self = Self(0x0B);
256
257	pub const JOIN_LEFT: Self = Self(0x0C);
258
259	pub const JOIN_RIGHT: Self = Self(0x0D);
260
261	pub const JOIN_SCHEMA: Self = Self(0x0E);
262
263	pub const RINGBUFFER_FORWARD: Self = Self(0x0F);
264
265	pub const RINGBUFFER_ENTRY: Self = Self(0x10);
266
267	pub const GATE_VISIBILITY: Self = Self(0x11);
268
269	pub const DISTINCT_LAYOUT: Self = Self(0x12);
270
271	pub const RINGBUFFER_EXPIRY: Self = Self(0x13);
272
273	pub const RINGBUFFER_TTL_ARM: Self = Self(0x14);
274
275	pub const SEAL_LEDGER: Self = Self(0x15);
276
277	pub const JOIN_PUBLISHED: Self = Self(0x16);
278
279	pub const JOIN_PIN: Self = Self(0x17);
280
281	pub const RINGBUFFER_META: Self = Self(0x18);
282
283	pub const REAP_QUEUE: Self = Self(0x19);
284
285	pub const JOIN_ROW_EXPIRY: Self = Self(0x1A);
286
287	pub const GUEST_ACCUMULATOR: Self = Self(0x1B);
288
289	pub const GUEST_BUFFER: Self = Self(0x1C);
290
291	pub const GUEST_RUNNING: Self = Self(0x1D);
292
293	pub const TUMBLING_EXPIRY: Self = Self(0x1E);
294
295	pub const PARTITIONED_RINGBUFFER_ENTRY: Self = Self(0x1F);
296
297	pub const PARTITIONED_RINGBUFFER_EXPIRY: Self = Self(0x20);
298
299	pub const PARTITIONED_RINGBUFFER_TTL_ARM: Self = Self(0x21);
300
301	pub const PARTITIONED_RINGBUFFER_META: Self = Self(0x22);
302
303	pub const CUSTOM_UNMANAGED: Self = Self(0x23);
304
305	pub const JOIN_EXPIRY_DUE: Self = Self(0x24);
306
307	pub const CUSTOM_MANAGED: Self = Self(0x25);
308
309	pub const CUSTOM_MANAGED_DUE: Self = Self(0x26);
310
311	pub const CUSTOM_MANAGED_LATEST: Self = Self(0x27);
312
313	pub const GUEST_WINDOW_PUBLISH: Self = Self(0x28);
314
315	pub const GUEST_RETAINED_ENTRY: Self = Self(0x29);
316
317	pub const LOOKUP_READ: Self = Self(0x2A);
318
319	pub const SESSION_MEMBER: Self = Self(0x2B);
320
321	pub const ROLLING_PANE: Self = Self(0x2D);
322
323	pub const TAKE_ROW: Self = Self(0x2E);
324
325	pub const RINGBUFFER_NEXT_ROW: Self = Self(0x2F);
326
327	pub fn name(&self) -> Cow<'static, str> {
328		match *self {
329			Self::NODE_COUNTER => "NODE_COUNTER",
330			Self::SOURCE_WATERMARK => "SOURCE_WATERMARK",
331			Self::TIMER_WHEEL => "TIMER_WHEEL",
332			Self::TIMER_INDEX => "TIMER_INDEX",
333			Self::JOIN_ROW_MAPPING => "JOIN_ROW_MAPPING",
334			Self::GROUP_ROW_MAPPING => "GROUP_ROW_MAPPING",
335			Self::GUEST_ROW_MAPPING => "GUEST_ROW_MAPPING",
336			Self::ACCUMULATOR => "ACCUMULATOR",
337			Self::BUFFER => "BUFFER",
338			Self::RUNNING => "RUNNING",
339			Self::EMIT => "EMIT",
340			Self::ROLLING_EXPIRY => "ROLLING_EXPIRY",
341			Self::COUNT => "COUNT",
342			Self::ROW_INDEX => "ROW_INDEX",
343			Self::SESSION => "SESSION",
344			Self::ROLLING_META => "ROLLING_META",
345			Self::ENGINE_META => "ENGINE_META",
346			Self::DISTINCT_ENTRY => "DISTINCT_ENTRY",
347			Self::WINDOW_META => "WINDOW_META",
348			Self::JOIN_LEFT => "JOIN_LEFT",
349			Self::JOIN_RIGHT => "JOIN_RIGHT",
350			Self::JOIN_SCHEMA => "JOIN_SCHEMA",
351			Self::RINGBUFFER_FORWARD => "RINGBUFFER_FORWARD",
352			Self::RINGBUFFER_ENTRY => "RINGBUFFER_ENTRY",
353			Self::GATE_VISIBILITY => "GATE_VISIBILITY",
354			Self::DISTINCT_LAYOUT => "DISTINCT_LAYOUT",
355			Self::RINGBUFFER_EXPIRY => "RINGBUFFER_EXPIRY",
356			Self::RINGBUFFER_TTL_ARM => "RINGBUFFER_TTL_ARM",
357			Self::SEAL_LEDGER => "SEAL_LEDGER",
358			Self::JOIN_PUBLISHED => "JOIN_PUBLISHED",
359			Self::JOIN_PIN => "JOIN_PIN",
360			Self::RINGBUFFER_META => "RINGBUFFER_META",
361			Self::REAP_QUEUE => "REAP_QUEUE",
362			Self::JOIN_ROW_EXPIRY => "JOIN_ROW_EXPIRY",
363			Self::JOIN_EXPIRY_DUE => "JOIN_EXPIRY_DUE",
364			Self::GUEST_ACCUMULATOR => "GUEST_ACCUMULATOR",
365			Self::GUEST_BUFFER => "GUEST_BUFFER",
366			Self::GUEST_RUNNING => "GUEST_RUNNING",
367			Self::TUMBLING_EXPIRY => "TUMBLING_EXPIRY",
368			Self::PARTITIONED_RINGBUFFER_ENTRY => "PARTITIONED_RINGBUFFER_ENTRY",
369			Self::PARTITIONED_RINGBUFFER_EXPIRY => "PARTITIONED_RINGBUFFER_EXPIRY",
370			Self::PARTITIONED_RINGBUFFER_TTL_ARM => "PARTITIONED_RINGBUFFER_TTL_ARM",
371			Self::PARTITIONED_RINGBUFFER_META => "PARTITIONED_RINGBUFFER_META",
372			Self::CUSTOM_UNMANAGED => "CUSTOM_UNMANAGED",
373			Self::CUSTOM_MANAGED => "CUSTOM_MANAGED",
374			Self::CUSTOM_MANAGED_DUE => "CUSTOM_MANAGED_DUE",
375			Self::CUSTOM_MANAGED_LATEST => "CUSTOM_MANAGED_LATEST",
376			Self::GUEST_WINDOW_PUBLISH => "GUEST_WINDOW_PUBLISH",
377			Self::GUEST_RETAINED_ENTRY => "GUEST_RETAINED_ENTRY",
378			Self::LOOKUP_READ => "LOOKUP_READ",
379			Self::SESSION_MEMBER => "SESSION_MEMBER",
380			Self::ROLLING_PANE => "ROLLING_PANE",
381			Self::TAKE_ROW => "TAKE_ROW",
382			Self::RINGBUFFER_NEXT_ROW => "RINGBUFFER_NEXT_ROW",
383			_ => return Cow::Owned(format!("{:#04x}", self.0)),
384		}
385		.into()
386	}
387
388	pub fn is_data(&self) -> bool {
389		self.0 <= Self::HIGHEST_DATA
390	}
391
392	pub fn is_identity(&self) -> bool {
393		!self.is_data()
394	}
395
396	pub const fn is_known(&self) -> bool {
397		REGISTERED[(self.0 >> 6) as usize] & (1u64 << (self.0 & 63)) != 0
398	}
399}
400
401pub fn is_framed_inner(inner: &[u8]) -> bool {
402	inner.is_empty() || OperatorStateKey::decode_inner(inner).is_some_and(|(_, keyspace, _)| keyspace.is_known())
403}
404
405const WINDOWED_KEYSPACES: [KeyspaceId; 12] = [
406	KeyspaceId::GUEST_ACCUMULATOR,
407	KeyspaceId::GUEST_BUFFER,
408	KeyspaceId::GUEST_RUNNING,
409	KeyspaceId::EMIT,
410	KeyspaceId::WINDOW_META,
411	KeyspaceId::SEAL_LEDGER,
412	KeyspaceId::ROLLING_EXPIRY,
413	KeyspaceId::TUMBLING_EXPIRY,
414	KeyspaceId::REAP_QUEUE,
415	KeyspaceId::GUEST_WINDOW_PUBLISH,
416	KeyspaceId::GUEST_RETAINED_ENTRY,
417	KeyspaceId::GUEST_ROW_MAPPING,
418];
419
420const LOWEST_META: u8 = KeyspaceId::GUEST_ROW_MAPPING.0;
421
422#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
423pub struct KeyspaceMask(u64);
424
425impl KeyspaceMask {
426	pub const fn all() -> Self {
427		Self(u64::MAX)
428	}
429
430	pub const fn from_bits(bits: u64) -> Self {
431		Self(bits)
432	}
433
434	pub fn windowed() -> Self {
435		Self::of(WINDOWED_KEYSPACES)
436	}
437
438	pub fn data() -> Self {
439		Self::of((0..=KeyspaceId::HIGHEST_DATA).map(KeyspaceId))
440	}
441
442	pub fn of(keyspaces: impl IntoIterator<Item = KeyspaceId>) -> Self {
443		let mut mask = Self::default();
444		for keyspace in keyspaces {
445			mask.insert(keyspace);
446		}
447		mask
448	}
449
450	pub fn holds(self, keyspace: KeyspaceId) -> bool {
451		keyspace_bit(keyspace).is_some_and(|bit| self.0 & bit != 0)
452	}
453
454	pub fn held(self) -> Vec<KeyspaceId> {
455		let mut ids: Vec<KeyspaceId> =
456			KEYSPACES.iter().map(|spec| spec.id).filter(|id| self.holds(*id)).collect();
457		ids.sort_unstable();
458		ids
459	}
460
461	pub fn insert(&mut self, keyspace: KeyspaceId) {
462		if let Some(bit) = keyspace_bit(keyspace) {
463			self.0 |= bit;
464		}
465	}
466
467	pub fn intersect(self, other: Self) -> Self {
468		Self(self.0 & other.0)
469	}
470
471	pub fn bits(self) -> u64 {
472		self.0
473	}
474}
475
476fn keyspace_bit(keyspace: KeyspaceId) -> Option<u64> {
477	let index = match keyspace.0 {
478		id if id <= KeyspaceId::HIGHEST_DATA => id as u32,
479		id if id >= LOWEST_META => (KeyspaceId::HIGHEST_DATA as u32) + 1 + (id - LOWEST_META) as u32,
480		_ => return None,
481	};
482	(index < u64::BITS).then(|| 1u64 << index)
483}
484
485pub fn guest_may_address(class: OperatorClass, keyspace: KeyspaceId) -> bool {
486	match class {
487		OperatorClass::Managed => keyspace == KeyspaceId::CUSTOM_MANAGED,
488		OperatorClass::Unmanaged => keyspace == KeyspaceId::CUSTOM_UNMANAGED,
489		OperatorClass::Nostate => false,
490		OperatorClass::Windowed => WINDOWED_KEYSPACES.contains(&keyspace),
491	}
492}
493
494pub fn is_class_framed_inner(class: OperatorClass, inner: &[u8]) -> bool {
495	OperatorStateKey::decode_inner(inner).is_some_and(|(_, keyspace, suffix)| {
496		guest_may_address(class, keyspace) && suffix_width_of(keyspace) == Some(suffix.len())
497	})
498}
499
500#[derive(Debug, Clone, PartialEq, KeyCodec, Eq, Hash)]
501#[key(tag = OperatorState)]
502pub struct OperatorStateKey {
503	pub operator: OperatorId,
504	pub group: GroupId,
505	pub keyspace: KeyspaceId,
506	pub suffix: Vec<u8>,
507}
508
509impl OperatorStateKey {
510	pub fn new(operator: OperatorId, group: GroupId, keyspace: KeyspaceId, suffix: impl Into<Vec<u8>>) -> Self {
511		Self {
512			operator,
513			group,
514			keyspace,
515			suffix: suffix.into(),
516		}
517	}
518
519	pub fn root(operator: OperatorId, keyspace: KeyspaceId, suffix: impl Into<Vec<u8>>) -> Self {
520		Self::new(operator, GroupId::ROOT, keyspace, suffix)
521	}
522
523	pub fn encoded(
524		operator: OperatorId,
525		group: GroupId,
526		keyspace: KeyspaceId,
527		suffix: impl AsRef<[u8]>,
528	) -> EncodedKey {
529		let suffix = suffix.as_ref();
530		let mut serializer = KeySerializer::with_capacity(NODE_GROUP_PREFIX_LEN + 1 + suffix.len());
531		serializer
532			.extend_u8(KeyTag::OperatorState as u8)
533			.extend_u64(operator.0)
534			.extend_fixed(*group.as_bytes())
535			.extend_u8(keyspace.0)
536			.extend_raw(suffix);
537		serializer.to_encoded_key()
538	}
539
540	pub fn inner(&self) -> EncodedKey {
541		let mut serializer = KeySerializer::with_capacity(KEYSPACE_INNER_PREFIX_LEN + self.suffix.len());
542		serializer.extend_fixed(*self.group.as_bytes()).extend_u8(self.keyspace.0).extend_raw(&self.suffix);
543		serializer.to_encoded_key()
544	}
545
546	pub const KEYSPACE_INNER_OFFSET: u32 = GroupId::WIDTH as u32;
547
548	pub fn decode_keyspace(stored: u8) -> KeyspaceId {
549		KeyspaceId(KeyDeserializer::from_bytes(&[stored]).read_u8().expect("a single byte decodes as u8"))
550	}
551
552	pub fn inner_encoded(group: GroupId, keyspace: KeyspaceId, suffix: impl AsRef<[u8]>) -> GroupStateKey {
553		let suffix = suffix.as_ref();
554		let mut serializer = KeySerializer::with_capacity(KEYSPACE_INNER_PREFIX_LEN + suffix.len());
555		serializer.extend_fixed(*group.as_bytes()).extend_u8(keyspace.0).extend_raw(suffix);
556		GroupStateKey(serializer.to_encoded_key())
557	}
558
559	pub fn decode_inner(inner: &[u8]) -> Option<(GroupId, KeyspaceId, &[u8])> {
560		let mut de = KeyDeserializer::from_bytes(inner);
561		let group = de.read_fixed().ok()?;
562		let keyspace = de.read_u8().ok()?;
563		let suffix = de.read_raw(de.remaining()).ok()?;
564		Some((GroupId::from_bytes(group), KeyspaceId(keyspace), suffix))
565	}
566
567	pub fn node_range(operator: OperatorId) -> TaggedKeyBoundRange {
568		node_range(operator)
569	}
570
571	pub fn decode_operator(key: &EncodedKey) -> Option<(OperatorId, EncodedKey)> {
572		let mut de = KeyDeserializer::from_bytes(key.as_slice());
573		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
574		if kind != KeyTag::OperatorState {
575			return None;
576		}
577		let operator = de.read_u64().ok()?;
578		let inner = de.read_raw(de.remaining()).ok()?;
579		Some((OperatorId(operator), EncodedKey::new(inner)))
580	}
581}
582
583#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
584pub struct GroupStateKey(EncodedKey);
585
586impl GroupStateKey {
587	pub fn new(group: GroupId, keyspace: KeyspaceId, suffix: impl AsRef<[u8]>) -> Self {
588		OperatorStateKey::inner_encoded(group, keyspace, suffix)
589	}
590
591	pub fn root(keyspace: KeyspaceId, suffix: impl AsRef<[u8]>) -> Self {
592		Self::new(GroupId::ROOT, keyspace, suffix)
593	}
594
595	pub fn from_framed(key: EncodedKey) -> Option<Self> {
596		is_framed_inner(key.as_slice()).then_some(Self(key))
597	}
598
599	pub fn from_class_framed(class: OperatorClass, key: EncodedKey) -> Option<Self> {
600		is_class_framed_inner(class, key.as_slice()).then_some(Self(key))
601	}
602
603	pub fn bound_unchecked(key: EncodedKey) -> Self {
604		Self(key)
605	}
606
607	pub fn as_encoded(&self) -> &EncodedKey {
608		&self.0
609	}
610
611	pub fn into_encoded(self) -> EncodedKey {
612		self.0
613	}
614
615	pub fn as_slice(&self) -> &[u8] {
616		self.0.as_slice()
617	}
618
619	pub fn as_bytes(&self) -> &[u8] {
620		self.0.as_bytes()
621	}
622
623	pub fn group(&self) -> Option<GroupId> {
624		OperatorStateKey::decode_inner(self.0.as_slice()).map(|(group, _, _)| group)
625	}
626
627	pub fn keyspace(&self) -> Option<KeyspaceId> {
628		let bytes = self.0.as_slice();
629		let offset = OperatorStateKey::KEYSPACE_INNER_OFFSET as usize;
630		(bytes.len() > offset).then(|| KeyspaceId(encode_u8(bytes[offset])))
631	}
632}
633
634#[derive(Debug, Clone, PartialEq, Eq)]
635pub enum GroupStateRange {
636	Keyspace {
637		group: GroupId,
638		keyspace: KeyspaceId,
639		start: Bound<Vec<u8>>,
640		end: Bound<Vec<u8>>,
641	},
642	Group(GroupId),
643	GroupData(GroupId),
644	GroupIdentity(GroupId),
645}
646
647impl GroupStateRange {
648	pub fn for_operator(self, operator: OperatorId) -> TaggedKeyBoundRange {
649		match self {
650			Self::Keyspace {
651				group,
652				keyspace,
653				start,
654				end,
655			} => {
656				let whole = keyspace_range(operator, group, keyspace);
657				let at = |suffix: &[u8], edge: SuffixEdge| {
658					TaggedKeyBound::Key(TaggedKey::OperatorState(OperatorStateKey::new(
659						operator,
660						group,
661						keyspace,
662						suffix_at_edge(keyspace, suffix, edge),
663					)))
664				};
665				let start = match start {
666					Bound::Unbounded => whole.start,
667					Bound::Included(suffix) => Bound::Included(at(&suffix, SuffixEdge::Low)),
668					Bound::Excluded(suffix) => Bound::Excluded(at(&suffix, SuffixEdge::High)),
669				};
670				let end = match end {
671					Bound::Unbounded => whole.end,
672					Bound::Included(suffix) => Bound::Included(at(&suffix, SuffixEdge::High)),
673					Bound::Excluded(suffix) => Bound::Excluded(at(&suffix, SuffixEdge::Low)),
674				};
675				TaggedKeyBoundRange {
676					start,
677					end,
678				}
679			}
680			Self::Group(group) => group_range(operator, group),
681			Self::GroupData(group) => group_data_range(operator, group),
682			Self::GroupIdentity(group) => group_identity_range(operator, group),
683		}
684	}
685}
686
687#[derive(Debug, Clone, PartialEq, Eq, Hash)]
688pub struct ManagedKey(GroupStateKey);
689
690impl AsRef<GroupStateKey> for ManagedKey {
691	fn as_ref(&self) -> &GroupStateKey {
692		&self.0
693	}
694}
695
696impl From<ManagedKey> for GroupStateKey {
697	fn from(key: ManagedKey) -> Self {
698		key.0
699	}
700}
701
702#[derive(Debug, Clone, PartialEq, Eq, Hash)]
703pub struct UnmanagedKey(GroupStateKey);
704
705impl AsRef<GroupStateKey> for UnmanagedKey {
706	fn as_ref(&self) -> &GroupStateKey {
707		&self.0
708	}
709}
710
711impl From<UnmanagedKey> for GroupStateKey {
712	fn from(key: UnmanagedKey) -> Self {
713		key.0
714	}
715}
716
717impl AsRef<[u8]> for GroupStateKey {
718	fn as_ref(&self) -> &[u8] {
719		self.0.as_slice()
720	}
721}
722
723impl AsRef<EncodedKey> for GroupStateKey {
724	fn as_ref(&self) -> &EncodedKey {
725		&self.0
726	}
727}
728
729pub trait IntoGroupStateKey {
730	fn into_group_state_key(self) -> GroupStateKey;
731}
732
733impl IntoGroupStateKey for GroupStateKey {
734	fn into_group_state_key(self) -> GroupStateKey {
735		self
736	}
737}
738
739fn group_inner_prefix(group: GroupId) -> Vec<u8> {
740	let mut serializer = KeySerializer::with_capacity(GroupId::WIDTH);
741	serializer.extend_fixed(*group.as_bytes());
742	serializer.finish().as_ref().to_vec()
743}
744
745fn keyspace_inner_prefix(group: GroupId, keyspace: KeyspaceId) -> Vec<u8> {
746	let mut prefix = group_inner_prefix(group);
747	prefix.push(encode_u8(keyspace.0));
748	prefix
749}
750
751pub fn group_inner_range(group: GroupId) -> EncodedKeyRange {
752	EncodedKeyRange::prefix(&group_inner_prefix(group))
753}
754
755pub fn keyspace_inner_range(group: GroupId, keyspace: KeyspaceId) -> EncodedKeyRange {
756	EncodedKeyRange::prefix(&keyspace_inner_prefix(group, keyspace))
757}
758
759enum SuffixEdge {
760	Low,
761	High,
762}
763
764fn suffix_at_edge(keyspace: KeyspaceId, suffix: &[u8], edge: SuffixEdge) -> Vec<u8> {
765	struct Pad<'a> {
766		suffix: &'a [u8],
767		edge: SuffixEdge,
768	}
769
770	impl KeyspaceVisitor for Pad<'_> {
771		type Output = Vec<u8>;
772
773		fn visit<K: Keyspace>(self) -> Self::Output {
774			let template = match self.edge {
775				SuffixEdge::Low => <K::Suffix as BoundedKey>::low().to_suffix_bytes(),
776				SuffixEdge::High => <K::Suffix as KeyLayout>::high().to_suffix_bytes(),
777			};
778			let mut bytes = self.suffix.to_vec();
779			bytes.truncate(template.len());
780			bytes.extend_from_slice(&template[bytes.len()..]);
781			bytes
782		}
783	}
784
785	dispatch(
786		keyspace,
787		Pad {
788			suffix,
789			edge,
790		},
791	)
792	.unwrap_or_else(|| suffix.to_vec())
793}
794
795pub type KeyspaceInnerRangeSplit = (GroupId, KeyspaceId, Bound<Vec<u8>>, Bound<Vec<u8>>);
796
797pub fn keyspace_inner_range_split(range: &EncodedKeyRange) -> Option<KeyspaceInnerRangeSplit> {
798	let (group, keyspace, start) = match &range.start {
799		Bound::Included(key) => {
800			let (group, keyspace, suffix) = OperatorStateKey::decode_inner(key.as_slice())?;
801			(group, keyspace, Bound::Included(suffix.to_vec()))
802		}
803		Bound::Excluded(key) => {
804			let (group, keyspace, suffix) = OperatorStateKey::decode_inner(key.as_slice())?;
805			(group, keyspace, Bound::Excluded(suffix.to_vec()))
806		}
807		Bound::Unbounded => return None,
808	};
809	let whole = EncodedKeyRange::prefix(&keyspace_inner_prefix(group, keyspace));
810	let end = if range.end == whole.end {
811		Bound::Unbounded
812	} else {
813		match &range.end {
814			Bound::Included(key) => match OperatorStateKey::decode_inner(key.as_slice())? {
815				(g, k, suffix) if g == group && k == keyspace => Bound::Included(suffix.to_vec()),
816				_ => return None,
817			},
818			Bound::Excluded(key) => match OperatorStateKey::decode_inner(key.as_slice())? {
819				(g, k, suffix) if g == group && k == keyspace => Bound::Excluded(suffix.to_vec()),
820				_ => return None,
821			},
822			Bound::Unbounded => return None,
823		}
824	};
825	Some((group, keyspace, start, end))
826}
827
828pub fn group_inner_range_split(range: &EncodedKeyRange) -> Option<GroupId> {
829	let key = match &range.start {
830		Bound::Included(key) | Bound::Excluded(key) => key,
831		Bound::Unbounded => return None,
832	};
833	let group = GroupId::from_bytes(KeyDeserializer::from_bytes(key.as_slice()).read_fixed().ok()?);
834	for candidate in [group_inner_range(group), group_data_inner_range(group), group_identity_inner_range(group)] {
835		if range.start == candidate.start && range.end == candidate.end {
836			return Some(group);
837		}
838	}
839	None
840}
841
842pub fn unmanaged_key_in(group: GroupId, id: &[u8]) -> Option<UnmanagedKey> {
843	CustomUnmanagedSuffix::of(id).map(|key| {
844		UnmanagedKey(OperatorStateKey::inner_encoded(
845			group,
846			KeyspaceId::CUSTOM_UNMANAGED,
847			key.to_suffix_bytes(),
848		))
849	})
850}
851
852pub fn unmanaged_key(id: &[u8]) -> Option<UnmanagedKey> {
853	unmanaged_key_in(GroupId::ROOT, id)
854}
855
856pub fn managed_key_in(group: GroupId, id: &[u8]) -> Option<ManagedKey> {
857	CustomManagedSuffix::of(id).map(|key| {
858		ManagedKey(OperatorStateKey::inner_encoded(group, KeyspaceId::CUSTOM_MANAGED, key.to_suffix_bytes()))
859	})
860}
861
862pub fn node_counter_key(kind: NodeCounterKind) -> GroupStateKey {
863	typed_key::<NodeCounter>(GroupId::ROOT, &NodeCounterKey::of(kind))
864}
865
866pub fn row_number_counter_key() -> GroupStateKey {
867	node_counter_key(NodeCounterKind::RowNumber)
868}
869
870pub fn group_data_inner_range(group: GroupId) -> EncodedKeyRange {
871	let prefix = group_inner_prefix(group);
872	let mut start = prefix.clone();
873	start.push(encode_u8(KeyspaceId::HIGHEST_DATA));
874	EncodedKeyRange::new(Bound::Included(EncodedKey::new(start)), EncodedKeyRange::prefix(&prefix).end)
875}
876
877pub fn group_identity_inner_range(group: GroupId) -> EncodedKeyRange {
878	let prefix = group_inner_prefix(group);
879	let mut end = prefix.clone();
880	end.push(encode_u8(KeyspaceId::HIGHEST_DATA));
881	EncodedKeyRange::new(Bound::Included(EncodedKey::new(prefix)), Bound::Excluded(EncodedKey::new(end)))
882}
883
884pub const NODE_PREFIX_LEN: usize = 9;
885
886pub const KEYSPACE_INNER_PREFIX_LEN: usize = GroupId::WIDTH + size_of::<u8>();
887
888pub const NODE_GROUP_PREFIX_LEN: usize = NODE_PREFIX_LEN + GroupId::WIDTH;
889
890pub fn extend_node_prefix(serializer: &mut KeySerializer, operator: OperatorId) {
891	serializer.extend_u8(KeyTag::OperatorState as u8).extend_u64(operator.0);
892}
893
894pub fn node_prefix(operator: OperatorId) -> Vec<u8> {
895	let mut serializer = KeySerializer::with_capacity(NODE_PREFIX_LEN);
896	extend_node_prefix(&mut serializer, operator);
897	serializer.finish().as_ref().to_vec()
898}
899
900pub fn inner_range(range: &TaggedKeyBoundRange) -> Option<(OperatorId, EncodedKeyRange)> {
901	let encoded = range.encode();
902	let start_key = match encoded.start.as_ref() {
903		Bound::Included(key) | Bound::Excluded(key) => key,
904		Bound::Unbounded => return None,
905	};
906	let (operator, _) = OperatorStateKey::decode_operator(start_key)?;
907	let prefix = node_prefix(operator);
908	let strip = |bound: Bound<&EncodedKey>| match bound {
909		Bound::Included(key) if key.as_slice().starts_with(&prefix) => {
910			Bound::Included(EncodedKey::new(&key.as_slice()[prefix.len()..]))
911		}
912		Bound::Excluded(key) if key.as_slice().starts_with(&prefix) => {
913			Bound::Excluded(EncodedKey::new(&key.as_slice()[prefix.len()..]))
914		}
915		_ => Bound::Unbounded,
916	};
917	Some((operator, EncodedKeyRange::new(strip(encoded.start.as_ref()), strip(encoded.end.as_ref()))))
918}
919
920pub fn node_range(operator: OperatorId) -> TaggedKeyBoundRange {
921	TaggedKeyBoundRange::prefix(KeyTag::OperatorState, [Field::UDesc(Width::U64, operator.0 as u128)])
922}
923
924pub fn group_range(operator: OperatorId, group: GroupId) -> TaggedKeyBoundRange {
925	TaggedKeyBoundRange::prefix(
926		KeyTag::OperatorState,
927		[
928			Field::UDesc(Width::U64, operator.0 as u128),
929			Field::BytesDesc(ByteEncoding::Fixed, Cow::Owned(group.as_bytes().to_vec())),
930		],
931	)
932}
933
934pub fn keyspace_range(operator: OperatorId, group: GroupId, keyspace: KeyspaceId) -> TaggedKeyBoundRange {
935	TaggedKeyBoundRange::prefix(
936		KeyTag::OperatorState,
937		[
938			Field::UDesc(Width::U64, operator.0 as u128),
939			Field::BytesDesc(ByteEncoding::Fixed, Cow::Owned(group.as_bytes().to_vec())),
940			Field::UDesc(Width::U8, keyspace.0 as u128),
941		],
942	)
943}
944
945pub fn group_data_range(operator: OperatorId, group: GroupId) -> TaggedKeyBoundRange {
946	TaggedKeyBoundRange {
947		start: Bound::Included(TaggedKeyBound::prefix(
948			KeyTag::OperatorState,
949			[
950				Field::UDesc(Width::U64, operator.0 as u128),
951				Field::BytesDesc(ByteEncoding::Fixed, Cow::Owned(group.as_bytes().to_vec())),
952				Field::UDesc(Width::U8, KeyspaceId::HIGHEST_DATA as u128),
953			],
954		)),
955		end: Bound::Excluded(TaggedKeyBound::prefix_end(
956			KeyTag::OperatorState,
957			[
958				Field::UDesc(Width::U64, operator.0 as u128),
959				Field::BytesDesc(ByteEncoding::Fixed, Cow::Owned(group.as_bytes().to_vec())),
960			],
961		)),
962	}
963}
964
965pub fn group_identity_range(operator: OperatorId, group: GroupId) -> TaggedKeyBoundRange {
966	TaggedKeyBoundRange {
967		start: Bound::Included(TaggedKeyBound::prefix(
968			KeyTag::OperatorState,
969			[
970				Field::UDesc(Width::U64, operator.0 as u128),
971				Field::BytesDesc(ByteEncoding::Fixed, Cow::Owned(group.as_bytes().to_vec())),
972			],
973		)),
974		end: Bound::Excluded(TaggedKeyBound::prefix(
975			KeyTag::OperatorState,
976			[
977				Field::UDesc(Width::U64, operator.0 as u128),
978				Field::BytesDesc(ByteEncoding::Fixed, Cow::Owned(group.as_bytes().to_vec())),
979				Field::UDesc(Width::U8, KeyspaceId::HIGHEST_DATA as u128),
980			],
981		)),
982	}
983}
984
985#[cfg(test)]
986mod tests {
987	use std::ops::Bound;
988
989	use reifydb_value::util::hash::Hash128;
990
991	use super::{
992		EncodedKey, EncodedKeyRange, GroupId, GroupSet, GroupStateKey, GroupStateRange, KeySerializer,
993		KeyspaceId, OperatorStateKey, group_data_inner_range, group_data_of_inner, group_data_range,
994		group_identity_inner_range, group_identity_range, group_inner_prefix, group_inner_range, group_range,
995		guest_may_address, is_class_framed_inner, is_framed_inner, keyspace_range, managed_key_in, node_prefix,
996		node_range, suffix_width_of, unmanaged_key_in,
997	};
998	use crate::{common::OperatorClass, interface::catalog::flow::OperatorId, key::operator::keyspace::KEYSPACES};
999
1000	const NODES: [u64; 4] = [1, 17, 300, 70_000];
1001	const GROUPS: [u128; 8] = [1, 2, 127, 128, 1000, 100_000, 1 << 30, u128::MAX];
1002	const DATA_KEYSPACES: [KeyspaceId; 5] = [
1003		KeyspaceId::ACCUMULATOR,
1004		KeyspaceId::BUFFER,
1005		KeyspaceId::RUNNING,
1006		KeyspaceId::CUSTOM_UNMANAGED,
1007		KeyspaceId::CUSTOM_MANAGED,
1008	];
1009	const IDENTITY_KEYSPACES: [KeyspaceId; 1] = [KeyspaceId::GUEST_ROW_MAPPING];
1010
1011	#[derive(Clone, Copy, PartialEq, Debug)]
1012	enum Phase {
1013		Data,
1014		Identity,
1015	}
1016
1017	/// Every keyspace the substrate declares, with the phase allowed to erase it and the tiers it may
1018	/// be cached in. Both are written down rather than read back from `is_data` and the `KEYSPACES` table, or
1019	/// a keyspace changing sides would pass unremarked.
1020	const CENSUS: [(&str, KeyspaceId, Phase, bool); 54] = [
1021		("NODE_COUNTER", KeyspaceId::NODE_COUNTER, Phase::Identity, true),
1022		("SOURCE_WATERMARK", KeyspaceId::SOURCE_WATERMARK, Phase::Identity, true),
1023		("TIMER_WHEEL", KeyspaceId::TIMER_WHEEL, Phase::Identity, true),
1024		("TIMER_INDEX", KeyspaceId::TIMER_INDEX, Phase::Identity, true),
1025		("JOIN_ROW_MAPPING", KeyspaceId::JOIN_ROW_MAPPING, Phase::Identity, true),
1026		("GROUP_ROW_MAPPING", KeyspaceId::GROUP_ROW_MAPPING, Phase::Identity, true),
1027		("GUEST_ROW_MAPPING", KeyspaceId::GUEST_ROW_MAPPING, Phase::Identity, true),
1028		("ACCUMULATOR", KeyspaceId::ACCUMULATOR, Phase::Data, true),
1029		("BUFFER", KeyspaceId::BUFFER, Phase::Data, true),
1030		("RUNNING", KeyspaceId::RUNNING, Phase::Data, true),
1031		("EMIT", KeyspaceId::EMIT, Phase::Data, true),
1032		("ROLLING_EXPIRY", KeyspaceId::ROLLING_EXPIRY, Phase::Data, true),
1033		("COUNT", KeyspaceId::COUNT, Phase::Data, true),
1034		("ROW_INDEX", KeyspaceId::ROW_INDEX, Phase::Data, true),
1035		("SESSION", KeyspaceId::SESSION, Phase::Data, true),
1036		("ROLLING_META", KeyspaceId::ROLLING_META, Phase::Data, true),
1037		("ENGINE_META", KeyspaceId::ENGINE_META, Phase::Data, true),
1038		("DISTINCT_ENTRY", KeyspaceId::DISTINCT_ENTRY, Phase::Data, true),
1039		("WINDOW_META", KeyspaceId::WINDOW_META, Phase::Data, true),
1040		("JOIN_LEFT", KeyspaceId::JOIN_LEFT, Phase::Data, true),
1041		("JOIN_RIGHT", KeyspaceId::JOIN_RIGHT, Phase::Data, true),
1042		("JOIN_SCHEMA", KeyspaceId::JOIN_SCHEMA, Phase::Data, true),
1043		("RINGBUFFER_FORWARD", KeyspaceId::RINGBUFFER_FORWARD, Phase::Data, true),
1044		("RINGBUFFER_ENTRY", KeyspaceId::RINGBUFFER_ENTRY, Phase::Data, true),
1045		("GATE_VISIBILITY", KeyspaceId::GATE_VISIBILITY, Phase::Data, true),
1046		("DISTINCT_LAYOUT", KeyspaceId::DISTINCT_LAYOUT, Phase::Data, true),
1047		("RINGBUFFER_EXPIRY", KeyspaceId::RINGBUFFER_EXPIRY, Phase::Data, true),
1048		("RINGBUFFER_TTL_ARM", KeyspaceId::RINGBUFFER_TTL_ARM, Phase::Data, true),
1049		("SEAL_LEDGER", KeyspaceId::SEAL_LEDGER, Phase::Data, true),
1050		("JOIN_PUBLISHED", KeyspaceId::JOIN_PUBLISHED, Phase::Data, true),
1051		("JOIN_PIN", KeyspaceId::JOIN_PIN, Phase::Data, true),
1052		("RINGBUFFER_META", KeyspaceId::RINGBUFFER_META, Phase::Data, true),
1053		("REAP_QUEUE", KeyspaceId::REAP_QUEUE, Phase::Data, true),
1054		("JOIN_ROW_EXPIRY", KeyspaceId::JOIN_ROW_EXPIRY, Phase::Data, true),
1055		("JOIN_EXPIRY_DUE", KeyspaceId::JOIN_EXPIRY_DUE, Phase::Data, true),
1056		("GUEST_ACCUMULATOR", KeyspaceId::GUEST_ACCUMULATOR, Phase::Data, true),
1057		("GUEST_BUFFER", KeyspaceId::GUEST_BUFFER, Phase::Data, true),
1058		("GUEST_RUNNING", KeyspaceId::GUEST_RUNNING, Phase::Data, true),
1059		("TUMBLING_EXPIRY", KeyspaceId::TUMBLING_EXPIRY, Phase::Data, true),
1060		("PARTITIONED_RINGBUFFER_ENTRY", KeyspaceId::PARTITIONED_RINGBUFFER_ENTRY, Phase::Data, true),
1061		("PARTITIONED_RINGBUFFER_EXPIRY", KeyspaceId::PARTITIONED_RINGBUFFER_EXPIRY, Phase::Data, true),
1062		("PARTITIONED_RINGBUFFER_TTL_ARM", KeyspaceId::PARTITIONED_RINGBUFFER_TTL_ARM, Phase::Data, true),
1063		("PARTITIONED_RINGBUFFER_META", KeyspaceId::PARTITIONED_RINGBUFFER_META, Phase::Data, true),
1064		("CUSTOM_UNMANAGED", KeyspaceId::CUSTOM_UNMANAGED, Phase::Data, false),
1065		("CUSTOM_MANAGED", KeyspaceId::CUSTOM_MANAGED, Phase::Data, false),
1066		("CUSTOM_MANAGED_DUE", KeyspaceId::CUSTOM_MANAGED_DUE, Phase::Data, true),
1067		("CUSTOM_MANAGED_LATEST", KeyspaceId::CUSTOM_MANAGED_LATEST, Phase::Data, true),
1068		("GUEST_WINDOW_PUBLISH", KeyspaceId::GUEST_WINDOW_PUBLISH, Phase::Data, true),
1069		("GUEST_RETAINED_ENTRY", KeyspaceId::GUEST_RETAINED_ENTRY, Phase::Data, true),
1070		("LOOKUP_READ", KeyspaceId::LOOKUP_READ, Phase::Data, true),
1071		("SESSION_MEMBER", KeyspaceId::SESSION_MEMBER, Phase::Data, true),
1072		("ROLLING_PANE", KeyspaceId::ROLLING_PANE, Phase::Data, true),
1073		("TAKE_ROW", KeyspaceId::TAKE_ROW, Phase::Data, true),
1074		("RINGBUFFER_NEXT_ROW", KeyspaceId::RINGBUFFER_NEXT_ROW, Phase::Data, true),
1075	];
1076
1077	/// Counts `KeyspaceId` constants from the source text. There is no reflection over associated
1078	/// constants, so this is the only way the census can notice a keyspace nobody listed.
1079	fn declared_keyspaces() -> usize {
1080		let source = include_str!("state.rs");
1081		let body = source
1082			.split("impl KeyspaceId {")
1083			.nth(1)
1084			.expect("the KeyspaceId impl block is where the constants are declared");
1085		let body = body.split("\n}\n").next().expect("the impl block is closed");
1086		body.lines()
1087			.filter(|line| {
1088				let line = line.trim_start();
1089				line.starts_with("pub const") && line.contains("Self(")
1090			})
1091			.count()
1092	}
1093
1094	#[test]
1095	fn a_bare_row_number_key_is_too_short_to_be_read_as_a_framed_key() {
1096		// a bare u64 is half a group prefix, so the framing check must decline it rather than read past its end
1097		let mut bare = KeySerializer::with_capacity(4);
1098		bare.extend_u64(7u64);
1099		let bare = bare.finish().as_ref().to_vec();
1100
1101		assert!(
1102			bare.len() < group_inner_prefix(GroupId::hashed(Hash128(7))).len(),
1103			"a bare row number cannot span a group"
1104		);
1105		assert!(OperatorStateKey::decode_inner(&bare).is_none());
1106		assert!(!is_framed_inner(&bare));
1107
1108		let framed = OperatorStateKey::inner_encoded(
1109			GroupId::ROOT,
1110			KeyspaceId::CUSTOM_UNMANAGED,
1111			7u64.to_be_bytes(),
1112		);
1113		assert!(is_framed_inner(framed.as_slice()));
1114		assert!(
1115			!contains(&group_identity_inner_range(GroupId::hashed(Hash128(7))), framed.as_slice()),
1116			"the framed form must sit outside every other group's range"
1117		);
1118	}
1119
1120	#[test]
1121	fn the_empty_key_is_framing_because_it_sorts_below_every_group() {
1122		// an empty inner key must sort below every group's prefix, or a reclaim phase could reach it
1123		let empty: &[u8] = &[];
1124		assert!(is_framed_inner(empty));
1125
1126		for group in GROUPS {
1127			let range = group_inner_range(GroupId::hashed(Hash128(group)));
1128			assert!(
1129				!contains(&range, empty),
1130				"the empty key must sit outside group {group}'s range, not merely be unattributed"
1131			);
1132		}
1133	}
1134
1135	#[test]
1136	fn the_empty_key_is_not_guest_framing_even_though_it_is_host_framing() {
1137		// the host reads an empty inner key as a sentinel sorting below every group, but a guest bound of
1138		// zero length used to inherit that and open a scan at the very bottom of the operator's keyspace,
1139		// sweeping every host keyspace on the way up; a guest key must name its keyspace or be refused
1140		let empty: &[u8] = &[];
1141		assert!(is_framed_inner(empty));
1142		assert!(!is_class_framed_inner(OperatorClass::Unmanaged, empty));
1143		assert!(GroupStateKey::from_class_framed(OperatorClass::Unmanaged, EncodedKey::new(Vec::new()))
1144			.is_none());
1145
1146		assert!(is_class_framed_inner(
1147			OperatorClass::Unmanaged,
1148			unmanaged_key_in(GroupId::hashed(Hash128(3)), &[])
1149				.expect("an empty id fits the keyspace")
1150				.as_ref()
1151				.as_slice()
1152		));
1153		assert!(
1154			!is_class_framed_inner(
1155				OperatorClass::Unmanaged,
1156				OperatorStateKey::inner_encoded(
1157					GroupId::hashed(Hash128(3)),
1158					KeyspaceId::CUSTOM_UNMANAGED,
1159					[]
1160				)
1161				.as_slice()
1162			),
1163			"a suffix narrower than its keyspace declares must be refused at the wall, or it reaches the 			 typed bucket and panics there instead"
1164		);
1165	}
1166
1167	#[test]
1168	fn a_keyspace_this_substrate_never_defines_is_not_framing() {
1169		// a two-byte group+keyspace pair must not be framing unless the keyspace is one the substrate declares
1170		let mut stray = KeySerializer::with_capacity(4);
1171		stray.extend_u64(3u64).extend_u8(0x90u8);
1172		assert!(!is_framed_inner(stray.finish().as_ref()));
1173
1174		for keyspace in DATA_KEYSPACES.iter().chain(IDENTITY_KEYSPACES.iter()) {
1175			assert!(
1176				is_framed_inner(
1177					OperatorStateKey::inner_encoded(GroupId::hashed(Hash128(3)), *keyspace, [])
1178						.as_slice()
1179				),
1180				"keyspace {keyspace:?} is one the substrate writes and must pass"
1181			);
1182		}
1183	}
1184
1185	fn contains(range: &EncodedKeyRange, key: &[u8]) -> bool {
1186		let after_start = match &range.start {
1187			Bound::Included(start) => key >= start.as_slice(),
1188			Bound::Excluded(start) => key > start.as_slice(),
1189			Bound::Unbounded => true,
1190		};
1191		let before_end = match &range.end {
1192			Bound::Included(end) => key <= end.as_slice(),
1193			Bound::Excluded(end) => key < end.as_slice(),
1194			Bound::Unbounded => true,
1195		};
1196		after_start && before_end
1197	}
1198
1199	fn population() -> Vec<OperatorStateKey> {
1200		let mut keys = Vec::new();
1201		for operator in NODES {
1202			for group in GROUPS {
1203				for keyspace in DATA_KEYSPACES.iter().chain(IDENTITY_KEYSPACES.iter()) {
1204					for coord in [0u64, 1, 999, u64::MAX] {
1205						keys.push(OperatorStateKey::new(
1206							OperatorId(operator),
1207							GroupId::hashed(Hash128(group)),
1208							*keyspace,
1209							coord.to_be_bytes().to_vec(),
1210						));
1211					}
1212				}
1213			}
1214			keys.push(OperatorStateKey::root(
1215				OperatorId(operator),
1216				KeyspaceId::NODE_COUNTER,
1217				b"7xKXtg2CW87d97TXJSDpbD5jBkheTqA83TZRuJosgAsU".to_vec(),
1218			));
1219		}
1220		keys
1221	}
1222
1223	#[test]
1224	fn a_group_range_contains_exactly_that_groups_keys() {
1225		// a group range must contain exactly that operator+group's keys, or reclaim destroys or leaks state
1226		let population = population();
1227		for operator in NODES {
1228			for group in GROUPS {
1229				let group_id = GroupId::hashed(Hash128(group));
1230				let range = group_range(OperatorId(operator), group_id).encode();
1231				for key in &population {
1232					let encoded = key.encode();
1233					let expected = key.operator.0 == operator && key.group == group_id;
1234					assert_eq!(
1235						contains(&range, encoded.as_slice()),
1236						expected,
1237						"operator {operator} group {group} range disagreed about a key of operator {} \
1238						 group {}",
1239						key.operator.0,
1240						key.group
1241					);
1242				}
1243			}
1244		}
1245	}
1246
1247	#[test]
1248	fn variable_length_group_ids_cannot_prefix_one_another() {
1249		// no group id's varint encoding may prefix another's, or reclaiming the shorter erases the longer's
1250		// keys
1251		let encodings: Vec<Vec<u8>> = GROUPS
1252			.iter()
1253			.map(|group| {
1254				OperatorStateKey::new(
1255					OperatorId(1),
1256					GroupId::hashed(Hash128(*group)),
1257					KeyspaceId::ACCUMULATOR,
1258					vec![],
1259				)
1260				.encode()
1261				.as_slice()
1262				.to_vec()
1263			})
1264			.collect();
1265
1266		for (i, a) in encodings.iter().enumerate() {
1267			for (j, b) in encodings.iter().enumerate() {
1268				if i != j {
1269					assert!(
1270						!b.starts_with(a.as_slice()),
1271						"group {} encodes as a prefix of group {}",
1272						GROUPS[i],
1273						GROUPS[j]
1274					);
1275				}
1276			}
1277		}
1278	}
1279
1280	#[test]
1281	fn the_data_and_identity_ranges_partition_the_group() {
1282		// data and identity keyspaces must fall in exactly one reclamation phase, never both or neither
1283		for operator in NODES {
1284			for group in GROUPS {
1285				let group = GroupId::hashed(Hash128(group));
1286				let data = group_data_range(OperatorId(operator), group).encode();
1287				let identity = group_identity_range(OperatorId(operator), group).encode();
1288
1289				for keyspace in DATA_KEYSPACES {
1290					let key = OperatorStateKey::new(
1291						OperatorId(operator),
1292						group,
1293						keyspace,
1294						vec![7, 7],
1295					)
1296					.encode();
1297					assert!(
1298						contains(&data, key.as_slice()),
1299						"data keyspace {keyspace:?} must fall in the phase-1 range"
1300					);
1301					assert!(
1302						!contains(&identity, key.as_slice()),
1303						"data keyspace {keyspace:?} must not fall in the phase-2 range"
1304					);
1305				}
1306
1307				for keyspace in IDENTITY_KEYSPACES {
1308					let key = OperatorStateKey::new(
1309						OperatorId(operator),
1310						group,
1311						keyspace,
1312						vec![7, 7],
1313					)
1314					.encode();
1315					assert!(
1316						contains(&identity, key.as_slice()),
1317						"identity keyspace {keyspace:?} must fall in the phase-2 range"
1318					);
1319					assert!(
1320						!contains(&data, key.as_slice()),
1321						"identity keyspace {keyspace:?} must survive phase 1"
1322					);
1323				}
1324			}
1325		}
1326	}
1327
1328	#[test]
1329	fn every_declared_keyspace_names_itself_for_offline_attribution() {
1330		// every declared keyspace must name itself, or an offline census misattributes it as CUSTOM
1331		for (name, keyspace, _, _) in CENSUS {
1332			assert_eq!(
1333				keyspace.name(),
1334				name,
1335				"{name} ({:#04x}) does not name itself, so an offline census reports it as CUSTOM",
1336				keyspace.0
1337			);
1338		}
1339
1340		assert_eq!(
1341			KeyspaceId::CUSTOM_UNMANAGED.name(),
1342			"CUSTOM_UNMANAGED",
1343			"a custom keyspace names the admission side it sits on; there is no unnamed fallback to absorb it"
1344		);
1345	}
1346
1347	#[test]
1348	fn every_declared_keyspace_states_whether_it_may_be_range_cached() {
1349		// A keyspace on the wrong side never gets its range cached, and the miss reads as a cold cache, not a
1350		// policy bug.
1351		for (name, keyspace, _, policy) in CENSUS {
1352			let range_cached =
1353				KEYSPACES.iter().find(|spec| spec.id == keyspace).map(|spec| spec.range_cached);
1354			assert_eq!(
1355				range_cached,
1356				Some(policy),
1357				"{name} ({:#04x}) is cached on a different side than the census records",
1358				keyspace.0
1359			);
1360		}
1361
1362		let uncached: Vec<&str> = CENSUS.iter().filter(|(_, _, _, p)| !*p).map(|(n, ..)| *n).collect();
1363		assert_eq!(
1364			uncached,
1365			["CUSTOM_UNMANAGED", "CUSTOM_MANAGED"],
1366			"widening the set the tier refuses turns that tier into an off switch and only shows up as a \
1367			 throughput loss in a replay, so every move in or out is a measured decision"
1368		);
1369	}
1370
1371	#[test]
1372	fn every_declared_keyspace_is_distinct_framing_and_swept_by_exactly_one_phase() {
1373		// every declared keyspace must have a unique byte and belong to exactly one reclamation phase
1374		assert_eq!(
1375			CENSUS.len(),
1376			declared_keyspaces(),
1377			"a keyspace was added to KeyspaceId without being added to the census, so nothing below \
1378			 ever looks at its byte"
1379		);
1380
1381		let mut seen: Vec<(&str, u8)> = Vec::new();
1382		for (name, keyspace, phase, _) in CENSUS {
1383			if let Some((other, _)) = seen.iter().find(|(_, byte)| *byte == keyspace.0) {
1384				panic!("{name} and {other} both claim keyspace byte {:#04x}", keyspace.0);
1385			}
1386			seen.push((name, keyspace.0));
1387
1388			assert!(
1389				keyspace.is_known(),
1390				"{name} is declared but not framing, so the sweep panics on the first row it holds"
1391			);
1392
1393			let group = GroupId::hashed(Hash128(4));
1394			let key = OperatorStateKey::new(OperatorId(9), group, keyspace, vec![7, 7]).encode();
1395			let data = contains(&group_data_range(OperatorId(9), group).encode(), key.as_slice());
1396			let identity = contains(&group_identity_range(OperatorId(9), group).encode(), key.as_slice());
1397
1398			assert!(data != identity, "{name} must fall in exactly one phase, not {data} and {identity}");
1399			assert_eq!(
1400				data,
1401				phase == Phase::Data,
1402				"{name} is declared {phase:?} but the phase-1 range says data={data}"
1403			);
1404			assert_eq!(
1405				keyspace.is_data(),
1406				phase == Phase::Data,
1407				"{name} is declared {phase:?} but is_data says {}",
1408				keyspace.is_data()
1409			);
1410		}
1411	}
1412
1413	#[test]
1414	fn root_entries_sit_outside_every_group_range() {
1415		// the root-scoped counter must sit outside every group range, or reclaiming a group erases it
1416		for operator in NODES {
1417			let counter = OperatorStateKey::root(
1418				OperatorId(operator),
1419				KeyspaceId::NODE_COUNTER,
1420				b"mint".to_vec(),
1421			)
1422			.encode();
1423			for group in GROUPS {
1424				let range = group_range(OperatorId(operator), GroupId::hashed(Hash128(group))).encode();
1425				assert!(
1426					!contains(&range, counter.as_slice()),
1427					"group {group} range must not contain the root group's counter"
1428				);
1429			}
1430		}
1431	}
1432
1433	#[test]
1434	fn a_node_range_contains_exactly_that_nodes_keys() {
1435		// a node range must contain exactly its own operator's keys, since drop_operator deletes by range
1436		let population = population();
1437		for operator in NODES {
1438			let range = node_range(OperatorId(operator)).encode();
1439			for key in &population {
1440				let encoded = key.encode();
1441				assert_eq!(
1442					contains(&range, encoded.as_slice()),
1443					key.operator.0 == operator,
1444					"operator {operator} range disagreed about a key of operator {}",
1445					key.operator.0
1446				);
1447			}
1448		}
1449	}
1450
1451	#[test]
1452	fn a_keyspace_range_isolates_one_keyspace_of_one_group() {
1453		// a keyspace range must isolate exactly one keyspace of one group, or scans mix incompatible payloads
1454		let operator = OperatorId(17);
1455		let group = GroupId::hashed(Hash128(42));
1456		let range = keyspace_range(operator, group, KeyspaceId::BUFFER).encode();
1457
1458		let inside = OperatorStateKey::new(operator, group, KeyspaceId::BUFFER, vec![1]).encode();
1459		assert!(contains(&range, inside.as_slice()));
1460
1461		for other in [KeyspaceId::ACCUMULATOR, KeyspaceId::RUNNING, KeyspaceId::GUEST_ROW_MAPPING] {
1462			let key = OperatorStateKey::new(operator, group, other, vec![1]).encode();
1463			assert!(!contains(&range, key.as_slice()), "keyspace {other:?} leaked into the buffer range");
1464		}
1465
1466		let other_group =
1467			OperatorStateKey::new(operator, GroupId::hashed(Hash128(43)), KeyspaceId::BUFFER, vec![1])
1468				.encode();
1469		assert!(!contains(&range, other_group.as_slice()), "another group's buffer leaked into the range");
1470	}
1471
1472	#[test]
1473	fn encode_decode_round_trips_every_component() {
1474		let key = OperatorStateKey::new(
1475			OperatorId(0xDEAD_BEEF),
1476			GroupId::hashed(Hash128(123_456)),
1477			KeyspaceId::CUSTOM_UNMANAGED,
1478			vec![1, 2, 3, 4],
1479		);
1480		assert_eq!(OperatorStateKey::decode(&key.encode()), Some(key));
1481	}
1482
1483	#[test]
1484	fn keys_still_decode_as_operator_state_of_their_node() {
1485		// every key of this kind must still decode to its own operator, or state is misrouted into the CDC log
1486		let key = OperatorStateKey::new(
1487			OperatorId(9),
1488			GroupId::hashed(Hash128(4)),
1489			KeyspaceId::ACCUMULATOR,
1490			vec![1],
1491		)
1492		.encode();
1493
1494		let decoded = OperatorStateKey::decode(&key).expect("must remain decodable as its key kind");
1495		assert_eq!(decoded.operator, OperatorId(9));
1496	}
1497
1498	#[test]
1499	fn an_inner_key_composed_with_its_node_prefix_reproduces_the_full_key() {
1500		// inner key plus node prefix must reproduce the full key, or state written through the API is
1501		// unreachable
1502		let key = OperatorStateKey::new(
1503			OperatorId(17),
1504			GroupId::hashed(Hash128(42)),
1505			KeyspaceId::BUFFER,
1506			vec![9, 9],
1507		);
1508
1509		let mut composed = node_prefix(OperatorId(17));
1510		composed.extend_from_slice(key.inner().as_slice());
1511
1512		assert_eq!(composed, key.encode().as_slice(), "inner key plus operator prefix must equal the full key");
1513	}
1514
1515	#[test]
1516	fn the_root_group_range_stays_inside_its_node() {
1517		// group 0's inner range has no byte-wise successor, so it must stay bounded by the operator prefix
1518		let range = GroupStateRange::Group(GroupId::ROOT).for_operator(OperatorId(17)).encode();
1519
1520		let own = OperatorStateKey::root(OperatorId(17), KeyspaceId::NODE_COUNTER, vec![1]).encode();
1521		assert!(contains(&range, own.as_slice()), "the operator's own counter must be in range");
1522
1523		for operator in NODES {
1524			if operator == 17 {
1525				continue;
1526			}
1527			for keyspace in [KeyspaceId::NODE_COUNTER, KeyspaceId::ACCUMULATOR] {
1528				let foreign =
1529					OperatorStateKey::new(OperatorId(operator), GroupId::ROOT, keyspace, vec![1])
1530						.encode();
1531				assert!(
1532					!contains(&range, foreign.as_slice()),
1533					"operator {operator} leaked into operator 17's root-group range"
1534				);
1535			}
1536		}
1537	}
1538
1539	#[test]
1540	fn inner_ranges_partition_the_group_like_their_full_key_counterparts() {
1541		// inner data/identity ranges must partition a group like their full-key counterparts, since reclamation
1542		// uses them
1543		let operator = OperatorId(17);
1544		for group in GROUPS {
1545			let group = GroupId::hashed(Hash128(group));
1546			let data = GroupStateRange::GroupData(group).for_operator(operator).encode();
1547			let identity = GroupStateRange::GroupIdentity(group).for_operator(operator).encode();
1548
1549			for keyspace in DATA_KEYSPACES {
1550				let key = OperatorStateKey::new(operator, group, keyspace, vec![7]).encode();
1551				assert!(contains(&data, key.as_slice()));
1552				assert!(!contains(&identity, key.as_slice()));
1553			}
1554			for keyspace in IDENTITY_KEYSPACES {
1555				let key = OperatorStateKey::new(operator, group, keyspace, vec![7]).encode();
1556				assert!(contains(&identity, key.as_slice()));
1557				assert!(!contains(&data, key.as_slice()));
1558			}
1559		}
1560	}
1561
1562	#[test]
1563	fn decode_inner_round_trips_the_tail() {
1564		let key = OperatorStateKey::new(
1565			OperatorId(3),
1566			GroupId::hashed(Hash128(77)),
1567			KeyspaceId::EMIT,
1568			vec![4, 5, 6],
1569		);
1570		let inner = key.inner();
1571		let (group, keyspace, suffix) =
1572			OperatorStateKey::decode_inner(inner.as_slice()).expect("inner must decode");
1573
1574		assert_eq!(group, GroupId::hashed(Hash128(77)));
1575		assert_eq!(keyspace, KeyspaceId::EMIT);
1576		assert_eq!(suffix, [4, 5, 6]);
1577	}
1578
1579	#[test]
1580	fn a_state_key_stays_compact_however_long_the_group_key_is() {
1581		// a hashed group id must keep the state key fixed-width, never embedding the raw group bytes
1582		let long = b"7xKXtg2CW87d97TXJSDpbD5jBkheTqA83TZRuJosgAsU\
1583		             So11111111111111111111111111111111111111112";
1584		let hashed = OperatorStateKey::new(
1585			OperatorId(17),
1586			GroupId::of(&EncodedKey::new(long)),
1587			KeyspaceId::ACCUMULATOR,
1588			vec![0; 8],
1589		)
1590		.encode();
1591		let short = OperatorStateKey::new(
1592			OperatorId(17),
1593			GroupId::of(&EncodedKey::new(b"g")),
1594			KeyspaceId::ACCUMULATOR,
1595			vec![0; 8],
1596		)
1597		.encode();
1598
1599		assert_eq!(hashed.as_slice().len(), short.as_slice().len(), "group key length must not reach the key");
1600		assert!(hashed.as_slice().len() * 2 < long.len() + 8, "and must stay far below embedding the bytes");
1601	}
1602
1603	#[test]
1604	fn the_ram_predicate_and_the_disk_range_agree_on_every_key() {
1605		// the RAM predicate and the disk range must agree on every key, or a phase-1 delete ghosts or strands a
1606		// row
1607		for group in GROUPS.map(|group| GroupId::hashed(Hash128(group))) {
1608			let range = group_data_inner_range(group);
1609			for other in GROUPS.map(|group| GroupId::hashed(Hash128(group))) {
1610				for keyspace in DATA_KEYSPACES.iter().chain(IDENTITY_KEYSPACES.iter()) {
1611					let key = OperatorStateKey::inner_encoded(other, *keyspace, vec![7, 7]);
1612					let in_range = contains(&range, key.as_slice());
1613					let in_predicate = group_data_of_inner(key.as_slice()) == Some(group);
1614					assert_eq!(
1615						in_range, in_predicate,
1616						"disk range and RAM predicate disagree for group {group:?} on a \
1617						 {keyspace:?} key of group {other:?}"
1618					);
1619				}
1620			}
1621		}
1622	}
1623
1624	#[test]
1625	fn the_ram_predicate_refuses_identity_keyspaces() {
1626		// the RAM predicate must never report an identity keyspace as reclaimable group data
1627		for keyspace in IDENTITY_KEYSPACES {
1628			let key = OperatorStateKey::inner_encoded(GroupId::hashed(Hash128(9)), keyspace, vec![1]);
1629			assert_eq!(
1630				group_data_of_inner(key.as_slice()),
1631				None,
1632				"{keyspace:?} must not be reported as reclaimable group data"
1633			);
1634		}
1635	}
1636
1637	#[test]
1638	fn a_key_too_short_to_carry_a_keyspace_is_refused() {
1639		// a key without both a group and a keyspace byte must not decode as group data
1640		assert_eq!(group_data_of_inner(&[]), None);
1641		assert_eq!(group_data_of_inner(&[0xAB]), None, "a group with no keyspace byte must not decode");
1642	}
1643
1644	#[test]
1645	fn the_predicate_agrees_with_the_disk_range_on_arbitrary_bytes() {
1646		// the predicate and the disk range must agree even on arbitrary bytes no encoder produced
1647		let mut seed = 0x2545F4914F6CDD1Du64;
1648		let mut next = move || {
1649			seed ^= seed << 13;
1650			seed ^= seed >> 7;
1651			seed ^= seed << 17;
1652			seed
1653		};
1654
1655		for _ in 0..2000 {
1656			let len = (next() % 12) as usize;
1657			let key: Vec<u8> = (0..len).map(|_| (next() % 256) as u8).collect();
1658			let Some(group) = group_data_of_inner(&key) else {
1659				continue;
1660			};
1661			assert!(
1662				contains(&group_data_inner_range(group), &key),
1663				"predicate attributed {key:?} to {group:?} but the disk range excludes it"
1664			);
1665		}
1666	}
1667
1668	#[test]
1669	fn a_group_set_is_sorted_deduped_and_never_admits_root() {
1670		// a group set must stay sorted and deduped for binary_search, and must never admit the root group
1671		let two = GroupId::hashed(Hash128(2));
1672		let five = GroupId::hashed(Hash128(5));
1673		let nine = GroupId::hashed(Hash128(9));
1674		let set = GroupSet::new([nine, two, nine, GroupId::ROOT, five]);
1675
1676		assert_eq!(set.as_slice(), &[two, five, nine]);
1677		assert_eq!(set.len(), 3);
1678		assert!(set.contains(five));
1679		assert!(!set.contains(GroupId::hashed(Hash128(3))));
1680		assert!(!set.contains(GroupId::ROOT), "the root group must be filtered out, not merely unsorted");
1681	}
1682
1683	#[test]
1684	fn an_empty_group_set_matches_nothing() {
1685		let set = GroupSet::new([]);
1686
1687		assert!(set.is_empty());
1688		assert!(!set.contains(GroupId::FIRST_NON_ROOT));
1689	}
1690
1691	#[test]
1692	fn a_managed_key_is_guest_framed_in_the_managed_keyspace() {
1693		// A managed key framed in any other keyspace is freed by the wrong owner, or never.
1694		let group = GroupId::hashed(Hash128(3));
1695		let key = managed_key_in(group, b"id").expect("a two byte id fits the keyspace");
1696		let (decoded, keyspace, _) =
1697			OperatorStateKey::decode_inner(key.as_ref().as_slice()).expect("a managed key decodes");
1698		assert_eq!(decoded, group);
1699		assert_eq!(keyspace, KeyspaceId::CUSTOM_MANAGED);
1700		assert!(is_class_framed_inner(OperatorClass::Managed, key.as_ref().as_slice()));
1701		assert!(
1702			managed_key_in(group, &[0u8; 17]).is_none(),
1703			"an id wider than the suffix must be refused, not cut"
1704		);
1705	}
1706
1707	#[test]
1708	fn each_class_addresses_exactly_the_keyspaces_it_owns() {
1709		// a keyspace admitted to the wrong class lets a guest write rows another owner frees, or no one does
1710		let owned = |class| {
1711			(0..=u8::MAX).map(KeyspaceId).filter(|id| guest_may_address(class, *id)).collect::<Vec<_>>()
1712		};
1713		assert_eq!(owned(OperatorClass::Managed), [KeyspaceId::CUSTOM_MANAGED]);
1714		assert_eq!(owned(OperatorClass::Unmanaged), [KeyspaceId::CUSTOM_UNMANAGED]);
1715		assert_eq!(owned(OperatorClass::Nostate), Vec::<KeyspaceId>::new());
1716		assert_eq!(
1717			owned(OperatorClass::Windowed),
1718			[
1719				KeyspaceId::EMIT,
1720				KeyspaceId::ROLLING_EXPIRY,
1721				KeyspaceId::WINDOW_META,
1722				KeyspaceId::SEAL_LEDGER,
1723				KeyspaceId::REAP_QUEUE,
1724				KeyspaceId::GUEST_ACCUMULATOR,
1725				KeyspaceId::GUEST_BUFFER,
1726				KeyspaceId::GUEST_RUNNING,
1727				KeyspaceId::TUMBLING_EXPIRY,
1728				KeyspaceId::GUEST_WINDOW_PUBLISH,
1729				KeyspaceId::GUEST_RETAINED_ENTRY,
1730				KeyspaceId::GUEST_ROW_MAPPING,
1731			]
1732		);
1733	}
1734
1735	#[test]
1736	fn a_class_framed_key_must_name_an_owned_keyspace_at_its_exact_width() {
1737		// a key framed for one class must never pass as another's, nor with a suffix its keyspace does not
1738		// declare
1739		let group = GroupId::hashed(Hash128(3));
1740		let at_width = |keyspace| {
1741			let width = suffix_width_of(keyspace).expect("every owned keyspace declares a width");
1742			OperatorStateKey::inner_encoded(group, keyspace, vec![0u8; width])
1743		};
1744		let managed = at_width(KeyspaceId::CUSTOM_MANAGED);
1745		assert!(is_class_framed_inner(OperatorClass::Managed, managed.as_slice()));
1746		for class in [OperatorClass::Unmanaged, OperatorClass::Nostate, OperatorClass::Windowed] {
1747			assert!(
1748				!is_class_framed_inner(class, managed.as_slice()),
1749				"{class:?} must not address a managed key"
1750			);
1751		}
1752		for keyspace in
1753			(0..=u8::MAX).map(KeyspaceId).filter(|id| guest_may_address(OperatorClass::Windowed, *id))
1754		{
1755			assert!(
1756				is_class_framed_inner(OperatorClass::Windowed, at_width(keyspace).as_slice()),
1757				"{keyspace:?}"
1758			);
1759		}
1760		let narrow = OperatorStateKey::inner_encoded(group, KeyspaceId::GUEST_ACCUMULATOR, []);
1761		assert!(!is_class_framed_inner(OperatorClass::Windowed, narrow.as_slice()));
1762	}
1763}
reifydb_core/key/operator/state.rs

reifydb_core/key/operator/
state.rs
