reifydb_core/key/
queue_deduplication.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::key::{
5	deserializer::KeyDeserializer,
6	encoded::{EncodedKey, EncodedKeyRange},
7	serializer::KeySerializer,
8};
9
10use super::{EncodableKey, KeyKind};
11use crate::interface::catalog::id::QueueId;
12
13#[derive(Debug, Clone, PartialEq)]
14pub struct QueueDeduplicationKey {
15	pub queue: QueueId,
16	pub key: Vec<u8>,
17}
18
19impl QueueDeduplicationKey {
20	pub fn new(queue: impl Into<QueueId>, key: impl Into<Vec<u8>>) -> Self {
21		Self {
22			queue: queue.into(),
23			key: key.into(),
24		}
25	}
26
27	pub fn encoded(queue: impl Into<QueueId>, key: impl Into<Vec<u8>>) -> EncodedKey {
28		Self::new(queue, key).encode()
29	}
30
31	pub fn full_scan(queue: QueueId) -> EncodedKeyRange {
32		EncodedKeyRange::start_end(Some(Self::scan_start(queue)), Some(Self::scan_end(queue)))
33	}
34
35	fn scan_start(queue: QueueId) -> EncodedKey {
36		let mut serializer = KeySerializer::with_capacity(9);
37		serializer.extend_u8(Self::KIND as u8).extend_u64(queue);
38		serializer.to_encoded_key()
39	}
40
41	fn scan_end(queue: QueueId) -> EncodedKey {
42		let mut serializer = KeySerializer::with_capacity(9);
43		serializer.extend_u8(Self::KIND as u8).extend_u64(*queue - 1);
44		serializer.to_encoded_key()
45	}
46}
47
48impl EncodableKey for QueueDeduplicationKey {
49	const KIND: KeyKind = KeyKind::QueueDeduplication;
50
51	fn encode(&self) -> EncodedKey {
52		let mut serializer = KeySerializer::with_capacity(9 + self.key.len() + 1);
53		serializer.extend_u8(Self::KIND as u8).extend_u64(self.queue).extend_bytes(&self.key);
54		serializer.to_encoded_key()
55	}
56
57	fn decode(key: &EncodedKey) -> Option<Self> {
58		let mut de = KeyDeserializer::from_bytes(key.as_slice());
59
60		let kind: KeyKind = de.read_u8().ok()?.try_into().ok()?;
61		if kind != Self::KIND {
62			return None;
63		}
64
65		let queue = de.read_u64().ok()?;
66		let deduplication = de.read_bytes().ok()?;
67
68		Some(Self {
69			queue: QueueId(queue),
70			key: deduplication,
71		})
72	}
73}
74
75#[cfg(test)]
76mod tests {
77	use std::ops::Bound;
78
79	use super::*;
80
81	#[test]
82	fn test_encode_decode_roundtrip() {
83		// A lossy codec either resurrects a claimed key or fails to recognise one, and both turn a
84		// duplicate enqueue into a second work item.
85		let encoded = QueueDeduplicationKey::encoded(QueueId(3), b"invoice-42".to_vec());
86		let decoded = QueueDeduplicationKey::decode(&encoded).unwrap();
87		assert_eq!(decoded.queue, QueueId(3));
88		assert_eq!(decoded.key, b"invoice-42".to_vec());
89	}
90
91	#[test]
92	fn test_arbitrary_bytes_survive_the_tail_encoding() {
93		// The tail is user-supplied, so it must survive the bytes the key codec treats as structural
94		// as well as embedded nul and multi-byte utf8; a mangled key dedups against the wrong record.
95		for key in [
96			vec![],
97			vec![0x00],
98			vec![0xff],
99			vec![0xff, 0x00, 0xff],
100			"order/\u{00e9}\u{4e2d}".as_bytes().to_vec(),
101		] {
102			let encoded = QueueDeduplicationKey::encoded(QueueId(1), key.clone());
103			let decoded = QueueDeduplicationKey::decode(&encoded).unwrap();
104			assert_eq!(decoded.key, key, "tail {key:?} must round-trip unchanged");
105		}
106	}
107
108	#[test]
109	fn test_the_same_key_in_two_queues_encodes_differently() {
110		// Two queues may legitimately use the same dedup key, so without the queue id discriminating,
111		// enqueueing "invoice-1" on one queue would suppress it on every other queue.
112		let a = QueueDeduplicationKey::encoded(QueueId(1), b"same".to_vec());
113		let b = QueueDeduplicationKey::encoded(QueueId(2), b"same".to_vec());
114		assert_ne!(a, b);
115	}
116
117	#[test]
118	fn test_full_scan_contains_only_the_target_queue() {
119		// Keys are stored bitwise-inverted, so a bound derived with the wrong sign makes the retention
120		// sweep either miss its own records or delete a neighbouring queue's.
121		let range = QueueDeduplicationKey::full_scan(QueueId(3));
122		let Bound::Included(start) = &range.start else {
123			panic!("expected an included start bound")
124		};
125		let Bound::Included(end) = &range.end else {
126			panic!("expected an included end bound")
127		};
128
129		assert!(start.as_slice() < end.as_slice(), "the range must be non-empty under byte order");
130
131		for key in [vec![], b"a".to_vec(), vec![0xff; 64]] {
132			let inside = QueueDeduplicationKey::encoded(QueueId(3), key.clone());
133			assert!(
134				inside.as_slice() >= start.as_slice() && inside.as_slice() <= end.as_slice(),
135				"key {key:?} in queue 3 must fall inside the scan range"
136			);
137		}
138
139		for queue in [QueueId(2), QueueId(4)] {
140			let neighbour = QueueDeduplicationKey::encoded(queue, b"a".to_vec());
141			assert!(
142				neighbour.as_slice() < start.as_slice() || neighbour.as_slice() > end.as_slice(),
143				"queue {queue:?} must fall outside queue 3's scan range"
144			);
145		}
146	}
147
148	#[test]
149	fn test_a_foreign_or_truncated_key_does_not_decode() {
150		// A partial record would collapse every key in the queue onto one dedup slot.
151		let encoded = QueueDeduplicationKey::encoded(QueueId(3), b"invoice-42".to_vec());
152
153		let mut wrong_kind = encoded.as_slice().to_vec();
154		wrong_kind[0] = KeyKind::Queue as u8;
155		assert_eq!(QueueDeduplicationKey::decode(&EncodedKey::new(wrong_kind)), None);
156
157		let truncated = encoded.as_slice()[..5].to_vec();
158		assert_eq!(QueueDeduplicationKey::decode(&EncodedKey::new(truncated)), None);
159	}
160}
reifydb_core/key/queue_deduplication.rs

reifydb_core/key/
queue_deduplication.rs
